You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by mx...@apache.org on 2016/11/06 09:04:37 UTC
[3/3] incubator-ariatosca git commit: ARIA-4 Create an API for
creating workflows
ARIA-4 Create an API for creating workflows
An API for creating workflows. Users can build graphs of tasks and
set depenedencies in between tasks to execute them in a specific
order.
This commit also includes minimal reorganization of a few test
modules, so they are now using the same file system hierarchy as
the modules which they test.
ARIA-5-Adapt-workflow-API-users-to-modified-API
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/017779dd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/017779dd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/017779dd
Branch: refs/heads/ARIA-3-api-for-creating-workflows
Commit: 017779dd564513714e67f8e6d6af5a8065736a48
Parents: e1c919d
Author: Ran Ziv <ra...@gigaspaces.com>
Authored: Fri Oct 21 13:21:25 2016 +0300
Committer: mxmrlv <mx...@gmail.com>
Committed: Sun Nov 6 11:03:02 2016 +0200
----------------------------------------------------------------------
aria/.pylintrc | 2 +-
aria/__init__.py | 2 +-
aria/cli/commands.py | 2 +-
aria/context/__init__.py | 20 +
aria/context/operation.py | 68 ++
aria/context/workflow.py | 188 +++++
aria/contexts.py | 211 ------
aria/decorators.py | 47 +-
aria/logger.py | 6 +-
aria/storage/models.py | 58 +-
aria/tools/lru_cache.py | 6 +-
aria/workflows/api/__init__.py | 2 +
aria/workflows/api/task.py | 90 +++
aria/workflows/api/task_graph.py | 294 ++++++++
aria/workflows/api/tasks_graph.py | 203 ------
aria/workflows/builtin/execute_operation.py | 34 +-
aria/workflows/builtin/heal.py | 167 +++--
aria/workflows/builtin/install.py | 23 +-
aria/workflows/builtin/uninstall.py | 17 +-
aria/workflows/builtin/workflows.py | 144 ++--
aria/workflows/core/__init__.py | 2 +
aria/workflows/core/executor.py | 3 +-
aria/workflows/core/task.py | 106 +++
aria/workflows/core/tasks.py | 104 ---
aria/workflows/core/translation.py | 45 +-
requirements.txt | 1 +
setup.py | 5 +
tests/.pylintrc | 6 +-
tests/mock/__init__.py | 16 +
tests/mock/context.py | 36 +
tests/mock/models.py | 132 ++++
tests/mock/operations.py | 33 +
tests/test_logger.py | 2 +-
tests/workflows/__init__.py | 2 +
tests/workflows/api/__init__.py | 0
tests/workflows/api/test_task.py | 120 ++++
tests/workflows/api/test_task_graph.py | 714 +++++++++++++++++++
tests/workflows/builtin/__init__.py | 53 ++
.../workflows/builtin/test_execute_operation.py | 74 ++
tests/workflows/builtin/test_heal.py | 117 +++
tests/workflows/builtin/test_install.py | 63 ++
tests/workflows/builtin/test_uninstall.py | 63 ++
tests/workflows/core/__init__.py | 0
tests/workflows/core/test_executor.py | 122 ++++
.../test_task_graph_into_exececution_graph.py | 83 +++
tests/workflows/test_executor.py | 117 ---
.../test_task_graph_into_exececution_graph.py | 79 --
tox.ini | 4 +-
48 files changed, 2693 insertions(+), 993 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/.pylintrc
----------------------------------------------------------------------
diff --git a/aria/.pylintrc b/aria/.pylintrc
index eb188a0..ee27fdb 100644
--- a/aria/.pylintrc
+++ b/aria/.pylintrc
@@ -62,7 +62,7 @@ confidence=
# --enable=similarities". If you want to run only the classes checker, but have
# no Warning level messages displayed, use"--disable=all --enable=classes
# --disable=W"
-disable=import-star-module-level,old-octal-literal,oct-method,print-statement,unpacking-in-except,parameter-unpacking,backtick,old-raise-syntax,old-ne-operator,long-suffix,dict-view-method,dict-iter-method,metaclass-assignment,next-method-called,raising-string,indexing-exception,raw_input-builtin,long-builtin,file-builtin,execfile-builtin,coerce-builtin,cmp-builtin,buffer-builtin,basestring-builtin,apply-builtin,filter-builtin-not-iterating,using-cmp-argument,useless-suppression,range-builtin-not-iterating,suppressed-message,no-absolute-import,old-division,cmp-method,reload-builtin,zip-builtin-not-iterating,intern-builtin,unichr-builtin,reduce-builtin,standarderror-builtin,unicode-builtin,xrange-builtin,coerce-method,delslice-method,getslice-method,setslice-method,input-builtin,round-builtin,hex-method,nonzero-method,map-builtin-not-iterating,redefined-builtin,logging-format-interpolation
+disable=import-star-module-level,old-octal-literal,oct-method,print-statement,unpacking-in-except,parameter-unpacking,backtick,old-raise-syntax,old-ne-operator,long-suffix,dict-view-method,dict-iter-method,metaclass-assignment,next-method-called,raising-string,indexing-exception,raw_input-builtin,long-builtin,file-builtin,execfile-builtin,coerce-builtin,cmp-builtin,buffer-builtin,basestring-builtin,apply-builtin,filter-builtin-not-iterating,using-cmp-argument,useless-suppression,range-builtin-not-iterating,suppressed-message,no-absolute-import,old-division,cmp-method,reload-builtin,zip-builtin-not-iterating,intern-builtin,unichr-builtin,reduce-builtin,standarderror-builtin,unicode-builtin,xrange-builtin,coerce-method,delslice-method,getslice-method,setslice-method,input-builtin,round-builtin,hex-method,nonzero-method,map-builtin-not-iterating,redefined-builtin,logging-format-interpolation,import-error
[REPORTS]
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/__init__.py
----------------------------------------------------------------------
diff --git a/aria/__init__.py b/aria/__init__.py
index 4d9b0f4..eca7b9b 100644
--- a/aria/__init__.py
+++ b/aria/__init__.py
@@ -56,7 +56,7 @@ def application_model_storage(driver):
models.DeploymentModification,
models.Execution,
models.ProviderContext,
- models.Operation,
+ models.Task,
])
return _model_storage[driver]
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/cli/commands.py
----------------------------------------------------------------------
diff --git a/aria/cli/commands.py b/aria/cli/commands.py
index a38229d..f909a1e 100644
--- a/aria/cli/commands.py
+++ b/aria/cli/commands.py
@@ -29,7 +29,7 @@ from aria import application_model_storage, application_resource_storage
from aria.logger import LoggerMixin
from aria.storage import FileSystemModelDriver, FileSystemResourceDriver
from aria.tools.application import StorageManager
-from aria.contexts import WorkflowContext
+from aria.context.workflow import WorkflowContext
from aria.workflows.core.engine import Engine
from aria.workflows.core.executor import ThreadExecutor
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/context/__init__.py
----------------------------------------------------------------------
diff --git a/aria/context/__init__.py b/aria/context/__init__.py
new file mode 100644
index 0000000..20e19db
--- /dev/null
+++ b/aria/context/__init__.py
@@ -0,0 +1,20 @@
+# 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.
+
+"""
+Provides contexts to workflow and operation
+"""
+
+from . import workflow, operation
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/context/operation.py
----------------------------------------------------------------------
diff --git a/aria/context/operation.py b/aria/context/operation.py
new file mode 100644
index 0000000..d4d229a
--- /dev/null
+++ b/aria/context/operation.py
@@ -0,0 +1,68 @@
+# 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.
+
+"""
+Workflow and operation contexts
+"""
+
+from uuid import uuid4
+
+from aria.logger import LoggerMixin
+
+class OperationContext(LoggerMixin):
+ """
+ Context object used during operation creation and execution
+ """
+
+ def __init__(
+ self,
+ name,
+ operation_details,
+ workflow_context,
+ node_instance,
+ inputs=None):
+ super(OperationContext, self).__init__()
+ self.name = name
+ self.id = str(uuid4())
+ self.operation_details = operation_details
+ self.workflow_context = workflow_context
+ self.node_instance = node_instance
+ self.inputs = inputs or {}
+
+ def __repr__(self):
+ details = ', '.join(
+ '{0}={1}'.format(key, value)
+ for key, value in self.operation_details.items())
+ return '{name}({0})'.format(details, name=self.name)
+
+ def __getattr__(self, attr):
+ try:
+ return getattr(self.workflow_context, attr)
+ except AttributeError:
+ return super(OperationContext, self).__getattribute__(attr)
+
+ @property
+ def operation(self):
+ """
+ The model operation
+ """
+ return self.storage.operation.get(self.id)
+
+ @operation.setter
+ def operation(self, value):
+ """
+ Store the operation in the model storage
+ """
+ self.storage.operation.store(value)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/context/workflow.py
----------------------------------------------------------------------
diff --git a/aria/context/workflow.py b/aria/context/workflow.py
new file mode 100644
index 0000000..b5b4e4b
--- /dev/null
+++ b/aria/context/workflow.py
@@ -0,0 +1,188 @@
+# 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.
+
+"""
+Workflow and operation contexts
+"""
+
+import threading
+from uuid import uuid4
+from contextlib import contextmanager
+
+from ..logger import LoggerMixin
+from ..tools.lru_cache import lru_cache
+
+
+class WorkflowContext(LoggerMixin):
+ """
+ Context object used during workflow creation and execution
+ """
+
+ def __init__(
+ self,
+ model_storage,
+ resource_storage,
+ deployment_id,
+ workflow_id,
+ # TODO: this should be changed when we decide where the execution is created
+ execution_id=None,
+ parameters=None,
+ **kwargs):
+ super(WorkflowContext, self).__init__(**kwargs)
+ self.id = str(uuid4())
+ self.model = model_storage
+ self.resource = resource_storage
+ self.deployment_id = deployment_id
+ self.workflow_id = workflow_id
+ self.execution_id = execution_id
+ self.parameters = parameters or {}
+
+ def __repr__(self):
+ return (
+ '{name}(deployment_id={self.deployment_id}, '
+ 'workflow_id={self.workflow_id}, '
+ 'execution_id={self.execution_id})'.format(
+ name=self.__class__.__name__, self=self))
+
+ @property
+ def blueprint_id(self):
+ """
+ The blueprint id
+ """
+ return self.deployment.blueprint_id
+
+ @property
+ @lru_cache()
+ def blueprint(self):
+ """
+ The blueprint model
+ """
+ return self.model.blueprint.get(self.blueprint_id)
+
+ @property
+ @lru_cache()
+ def deployment(self):
+ """
+ The deployment model
+ """
+ return self.model.deployment.get(self.deployment_id)
+
+ @property
+ def nodes(self):
+ """
+ Iterator over nodes
+ """
+ return self.model.node.iter(
+ filters={'blueprint_id': self.blueprint_id})
+
+ @property
+ def node_instances(self):
+ """
+ Iterator over node instances
+ """
+ return self.model.node_instance.iter(filters={'deployment_id': self.deployment_id})
+
+ @property
+ def execution(self):
+ """
+ The execution model
+ """
+ return self.model.execution.get(self.execution_id)
+
+ @execution.setter
+ def execution(self, value):
+ """
+ Store the execution in the model storage
+ """
+ self.model.execution.store(value)
+
+ def download_blueprint_resource(self, destination, path=None):
+ """
+ Download a blueprint resource from the resource storage
+ """
+ return self.resource.blueprint.download(
+ entry_id=self.blueprint_id,
+ destination=destination,
+ path=path)
+
+ def download_deployment_resource(self, destination, path=None):
+ """
+ Download a deployment resource from the resource storage
+ """
+ return self.resource.deployment.download(
+ entry_id=self.deployment_id,
+ destination=destination,
+ path=path)
+
+ @lru_cache()
+ def get_deployment_resource_data(self, path=None):
+ """
+ Read a deployment resource as string from the resource storage
+ """
+ return self.resource.deployment.data(entry_id=self.deployment_id, path=path)
+
+ @lru_cache()
+ def get_blueprint_resource_data(self, path=None):
+ """
+ Read a blueprint resource as string from the resource storage
+ """
+ return self.resource.blueprint.data(entry_id=self.blueprint_id, path=path)
+
+
+class _CurrentContext(threading.local):
+ """
+ Provides thread-level context, which sugarcoats the task api.
+ """
+
+ def _set(self, value):
+ setattr(self, '_workflow_context', value)
+
+ def get(self):
+ """
+ Retrieves the current workflow context
+ :return: the workflow context
+ :rtype: WorkflowContext
+ """
+ try:
+ return getattr(self, '_workflow_context')
+ except AttributeError:
+ return None
+
+ def clear(self):
+ """
+ Clears the current context
+ """
+ if hasattr(self, '_workflow_context'):
+ delattr(self, '_workflow_context')
+
+ @contextmanager
+ def push(self, workflow_context):
+ """
+ Switches the current context to the provides context
+ :param workflow_context: the context to switch to.
+ :yields: the current context
+ """
+ if hasattr(self, '_workflow_context'):
+ prev_workflow_context = getattr(self, '_workflow_context')
+ else:
+ prev_workflow_context = None
+ self._set(workflow_context)
+ try:
+ yield self
+ finally:
+ self._set(prev_workflow_context)
+
+current = _CurrentContext()
+
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/contexts.py
----------------------------------------------------------------------
diff --git a/aria/contexts.py b/aria/contexts.py
deleted file mode 100644
index ae7fc66..0000000
--- a/aria/contexts.py
+++ /dev/null
@@ -1,211 +0,0 @@
-# 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.
-
-"""
-Workflow and operation contexts
-"""
-
-from uuid import uuid4
-
-from aria.logger import LoggerMixin
-from aria.tools.lru_cache import lru_cache
-from aria.workflows.api.tasks_graph import TaskGraph
-
-
-class WorkflowContext(LoggerMixin):
- """
- Context object used during workflow creation and execution
- """
-
- def __init__(
- self,
- name,
- model_storage,
- resource_storage,
- deployment_id,
- workflow_id,
- parameters=None,
- **kwargs):
- super(WorkflowContext, self).__init__(**kwargs)
- self.name = name
- self.id = str(uuid4())
- self.model = model_storage
- self.resource = resource_storage
- self.deployment_id = deployment_id
- self.workflow_id = workflow_id
- self.execution_id = str(uuid4())
- self.parameters = parameters or {}
-
- def __repr__(self):
- return (
- '{name}(deployment_id={self.deployment_id}, '
- 'workflow_id={self.workflow_id}, '
- 'execution_id={self.execution_id})'.format(
- name=self.__class__.__name__, self=self))
-
- def operation(
- self,
- name,
- operation_details,
- node_instance,
- inputs=None):
- """
- Called during workflow creation, return an operation context. This object should be added to
- the task graph.
- """
- return OperationContext(
- name=name,
- operation_details=operation_details,
- workflow_context=self,
- node_instance=node_instance,
- inputs=inputs or {})
-
- @property
- def task_graph(self):
- """
- The task graph class
- """
- return TaskGraph
-
- @property
- def blueprint_id(self):
- """
- The blueprint id
- """
- return self.deployment.blueprint_id
-
- @property
- @lru_cache()
- def blueprint(self):
- """
- The blueprint model
- """
- return self.model.blueprint.get(self.blueprint_id)
-
- @property
- @lru_cache()
- def deployment(self):
- """
- The deployment model
- """
- return self.model.deployment.get(self.deployment_id)
-
- @property
- def nodes(self):
- """
- Iterator over nodes
- """
- return self.model.node.iter(
- filters={'blueprint_id': self.blueprint_id})
-
- @property
- def node_instances(self):
- """
- Iterator over node instances
- """
- return self.model.node_instance.iter(filters={'deployment_id': self.deployment_id})
-
- @property
- def execution(self):
- """
- The execution model
- """
- return self.model.execution.get(self.execution_id)
-
- @execution.setter
- def execution(self, value):
- """
- Store the execution in the model storage
- """
- self.model.execution.store(value)
-
- def download_blueprint_resource(self, destination, path=None):
- """
- Download a blueprint resource from the resource storage
- """
- return self.resource.blueprint.download(
- entry_id=self.blueprint_id,
- destination=destination,
- path=path)
-
- def download_deployment_resource(self, destination, path=None):
- """
- Download a deployment resource from the resource storage
- """
- return self.resource.deployment.download(
- entry_id=self.deployment_id,
- destination=destination,
- path=path)
-
- @lru_cache()
- def get_deployment_resource_data(self, path=None):
- """
- Read a deployment resource as string from the resource storage
- """
- return self.resource.deployment.data(entry_id=self.deployment_id, path=path)
-
- @lru_cache()
- def get_blueprint_resource_data(self, path=None):
- """
- Read a blueprint resource as string from the resource storage
- """
- return self.resource.blueprint.data(entry_id=self.blueprint_id, path=path)
-
-
-class OperationContext(LoggerMixin):
- """
- Context object used during operation creation and execution
- """
-
- def __init__(
- self,
- name,
- operation_details,
- workflow_context,
- node_instance,
- inputs=None):
- super(OperationContext, self).__init__()
- self.name = name
- self.id = str(uuid4())
- self.operation_details = operation_details
- self.workflow_context = workflow_context
- self.node_instance = node_instance
- self.inputs = inputs or {}
-
- def __repr__(self):
- details = ', '.join(
- '{0}={1}'.format(key, value)
- for key, value in self.operation_details.items())
- return '{name}({0})'.format(details, name=self.name)
-
- def __getattr__(self, attr):
- try:
- return getattr(self.workflow_context, attr)
- except AttributeError:
- return super(OperationContext, self).__getattribute__(attr)
-
- @property
- def operation(self):
- """
- The model operation
- """
- return self.storage.operation.get(self.id)
-
- @operation.setter
- def operation(self, value):
- """
- Store the operation in the model storage
- """
- self.storage.operation.store(value)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/decorators.py
----------------------------------------------------------------------
diff --git a/aria/decorators.py b/aria/decorators.py
index 7bc41b3..5713246 100644
--- a/aria/decorators.py
+++ b/aria/decorators.py
@@ -20,12 +20,13 @@ Workflow and operation decorators
from uuid import uuid4
from functools import partial, wraps
-from aria.tools.validation import validate_function_arguments
+from . import context
+from .workflows.api import task_graph
+from .tools.validation import validate_function_arguments
def workflow(
func=None,
- workflow_context=True,
simple_workflow=True,
suffix_template=''):
"""
@@ -34,31 +35,29 @@ def workflow(
if func is None:
return partial(
workflow,
- workflow_context=workflow_context,
simple_workflow=simple_workflow,
suffix_template=suffix_template)
@wraps(func)
- def _wrapper(context, **custom_kwargs):
+ def _wrapper(ctx, **custom_kwargs):
+
workflow_name = _generate_workflow_name(
func_name=func.__name__,
suffix_template=suffix_template,
- context=context,
+ ctx=ctx,
**custom_kwargs)
- func_kwargs = _create_func_kwargs(
- custom_kwargs,
- context,
- add_context=workflow_context,
- workflow_name=workflow_name)
- validate_function_arguments(func, func_kwargs)
- func(**func_kwargs)
- return func_kwargs['graph']
+
+ custom_kwargs.setdefault('ctx', ctx)
+ custom_kwargs.setdefault('graph', task_graph.TaskGraph(workflow_name))
+ validate_function_arguments(func, custom_kwargs)
+ with context.workflow.current.push(ctx):
+ func(**custom_kwargs)
+ return custom_kwargs['graph']
return _wrapper
def operation(
- func=None,
- operation_context=True):
+ func=None):
"""
Operation decorator
"""
@@ -66,29 +65,25 @@ def operation(
return partial(operation)
@wraps(func)
- def _wrapper(context, **custom_kwargs):
+ def _wrapper(ctx, **custom_kwargs):
func_kwargs = _create_func_kwargs(
custom_kwargs,
- context,
- add_context=operation_context)
+ ctx)
validate_function_arguments(func, func_kwargs)
- context.description = func.__doc__
+ ctx.description = func.__doc__
return func(**func_kwargs)
return _wrapper
-def _generate_workflow_name(func_name, context, suffix_template, **custom_kwargs):
+def _generate_workflow_name(func_name, ctx, suffix_template, **custom_kwargs):
return '{func_name}.{suffix}'.format(
func_name=func_name,
- suffix=suffix_template.format(context=context, **custom_kwargs) or str(uuid4()))
+ suffix=suffix_template.format(ctx=ctx, **custom_kwargs) or str(uuid4()))
def _create_func_kwargs(
kwargs,
- context,
- add_context=True,
+ ctx,
workflow_name=None):
- if add_context:
- kwargs['context'] = context
- kwargs.setdefault('graph', context.task_graph(workflow_name))
+ kwargs.setdefault('graph', ctx.task_graph(workflow_name))
return kwargs
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/logger.py
----------------------------------------------------------------------
diff --git a/aria/logger.py b/aria/logger.py
index 4959126..0002cb5 100644
--- a/aria/logger.py
+++ b/aria/logger.py
@@ -37,7 +37,7 @@ class LoggerMixin(object):
def __init__(self, *args, **kwargs):
self.logger_name = self.logger_name or self.__class__.__name__
- self.logger = _base_logger.getChild(self.logger_name)
+ self.logger = logging.getLogger('{0}.{1}'.format(_base_logger.name, self.logger_name))
self.logger.setLevel(self.logger_level)
super(LoggerMixin, self).__init__(*args, **kwargs)
@@ -63,7 +63,7 @@ class LoggerMixin(object):
def __setstate__(self, obj_dict):
vars(self).update(
- logger=_base_logger.getChild(obj_dict['logger_name']),
+ logger=logging.getLogger('{0}.{1}'.format(_base_logger.name, obj_dict['logger_name'])),
**obj_dict)
@@ -112,7 +112,7 @@ class _DefaultConsoleFormat(logging.Formatter):
self._fmt = '%(levelname)s: %(message)s'
except AttributeError:
return record.message
- return super(_DefaultConsoleFormat, self).format(record)
+ return logging.Formatter.format(self, record)
def create_file_log_handler(
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/storage/models.py
----------------------------------------------------------------------
diff --git a/aria/storage/models.py b/aria/storage/models.py
index 32403ed..64c00bc 100644
--- a/aria/storage/models.py
+++ b/aria/storage/models.py
@@ -216,32 +216,6 @@ class Execution(Model):
is_system_workflow = Field(type=bool, default=False)
-class Operation(Model):
- """
- A Model which represents an operation
- """
- PENDING = 'pending'
- STARTED = 'started'
- SUCCESS = 'success'
- FAILED = 'failed'
- STATES = (
- PENDING,
- STARTED,
- SUCCESS,
- FAILED,
- )
- END_STATES = [SUCCESS, FAILED]
-
- id = Field(type=basestring, default=uuid_generator)
- status = Field(type=basestring, choices=STATES, default=STARTED)
- execution_id = Field(type=basestring)
- eta = Field(type=datetime, default=0)
- started_at = Field(type=datetime, default=None)
- ended_at = Field(type=datetime, default=None)
- max_retries = Field(type=int, default=0)
- retry_count = Field(type=int, default=0)
-
-
class Relationship(Model):
"""
A Model which represents a relationship
@@ -395,3 +369,35 @@ class Plugin(Model):
excluded_wheels = Field()
supported_py_versions = Field(type=list)
uploaded_at = Field(type=datetime)
+
+
+class Task(Model):
+ """
+ A Model which represents an task
+ """
+ PENDING = 'pending'
+ STARTED = 'started'
+ SUCCESS = 'success'
+ FAILED = 'failed'
+ STATES = (
+ PENDING,
+ STARTED,
+ SUCCESS,
+ FAILED,
+ )
+ END_STATES = [SUCCESS, FAILED]
+
+ id = Field(type=basestring, default=uuid_generator)
+ status = Field(type=basestring, choices=STATES, default=STARTED)
+ execution_id = Field(type=basestring)
+ eta = Field(type=datetime, default=0)
+ started_at = Field(type=datetime, default=None)
+ ended_at = Field(type=datetime, default=None)
+ max_retries = Field(type=int, default=0)
+ retry_count = Field(type=int, default=0)
+
+ # Operation specific fields
+ name = Field(type=basestring)
+ operation_details = Field(type=dict)
+ node_instance = PointerField(type=NodeInstance)
+ inputs = Field(type=dict, default=lambda: {})
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/tools/lru_cache.py
----------------------------------------------------------------------
diff --git a/aria/tools/lru_cache.py b/aria/tools/lru_cache.py
index 5863376..bb39b90 100755
--- a/aria/tools/lru_cache.py
+++ b/aria/tools/lru_cache.py
@@ -21,7 +21,11 @@ Function lru_cache implementation for python 2.7
from time import time
from functools import partial, wraps
from itertools import imap
-from collections import OrderedDict
+
+try:
+ from collections import OrderedDict
+except ImportError:
+ from ordereddict import OrderedDict
class _LRUCache(object):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/__init__.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/__init__.py b/aria/workflows/api/__init__.py
index ae1e83e..b7f4b0c 100644
--- a/aria/workflows/api/__init__.py
+++ b/aria/workflows/api/__init__.py
@@ -12,3 +12,5 @@
# 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 . import task, task_graph
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/task.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/task.py b/aria/workflows/api/task.py
new file mode 100644
index 0000000..a24fdc4
--- /dev/null
+++ b/aria/workflows/api/task.py
@@ -0,0 +1,90 @@
+# 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.
+
+"""
+Provides the tasks to be enterd into the task graph
+"""
+from uuid import uuid4
+
+from aria import context
+
+
+class BaseTask(object):
+ """
+ Abstract task_graph task
+ """
+ pass
+
+
+class OperationTask(BaseTask):
+ """
+ Represents an operation task in the task_graph
+ """
+
+ def __init__(self,
+ name,
+ operation_details,
+ node_instance,
+ **operation_kwargs):
+ """
+ Creates an operation task using the name, details, node instance and any additional kwargs.
+ :param name: the operation of the name.
+ :param operation_details: the details for the operation.
+ :param node_instance: the node instnace on which this operation is registered.
+ :param operation_kwargs: any additional kwargs for the operation.
+ """
+ self.id = uuid4()
+ self.workflow_context = context.workflow.current.get()
+ self.name = name
+ self.operation_details = operation_details
+ self.node_instance = node_instance
+ self.operation_kwargs = operation_kwargs
+
+ def __getattr__(self, item):
+ try:
+ return self.__getattribute__(item)
+ except AttributeError:
+ return getattr(self._workflow_context.model.task.get(self._task_model.id), item)
+
+
+class WorkflowTask(BaseTask):
+ """
+ Represents an workflow task in the task_graph
+ """
+ def __init__(self, workflow_func, **kwargs):
+ """
+ Creates a workflow task using the workflow_func provided and its kwargs
+ :param workflow_func: the function to run
+ :param kwargs: the kwargs that would be passed to the workflow_func
+ """
+ kwargs.setdefault('ctx', context.workflow.current.get())
+ self._graph = workflow_func(**kwargs)
+
+ @property
+ def graph(self):
+ """
+ The graph constructed by the sub workflow
+ :return:
+ """
+ return self._graph
+
+ def __getattr__(self, item):
+ try:
+ return self.__getattribute__(item)
+ except AttributeError:
+ try:
+ return getattr(self._graph, item)
+ except AttributeError:
+ return getattr(context.workflow.current.get(), item)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/task_graph.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/task_graph.py b/aria/workflows/api/task_graph.py
new file mode 100644
index 0000000..42cf8b6
--- /dev/null
+++ b/aria/workflows/api/task_graph.py
@@ -0,0 +1,294 @@
+# 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.
+
+"""
+Task graph. Used by users to build workflows
+"""
+
+from uuid import uuid4
+from collections import Iterable
+
+from networkx import DiGraph, topological_sort
+
+from . import task
+
+
+class TaskNotInGraphError(Exception):
+ """
+ An error representing a scenario where a given task is not in the graph as expected
+ """
+ pass
+
+
+def _filter_out_empty_tasks(func=None):
+ if func is None:
+ return lambda f: _filter_out_empty_tasks(func=f)
+
+ def _wrapper(task, *tasks, **kwargs):
+ tasks = filter(lambda t: bool(t), [task] + list(tasks))
+ return func(*tasks, **kwargs)
+ return _wrapper
+
+
+class TaskGraph(object):
+ """
+ A tasks graph builder.
+ Build an operations flow graph
+ """
+
+ def __init__(self, name):
+ self.name = name
+ self._id = str(uuid4())
+ self._graph = DiGraph()
+
+ def __repr__(self):
+ return '{name}(id={self._id}, name={self.name}, graph={self._graph!r})'.format(
+ name=self.__class__.__name__, self=self)
+
+ @property
+ def id(self):
+ """
+ Represents the id of the graph
+ :return: graph id
+ """
+ return self._id
+
+ # graph traversal methods
+
+ @property
+ def tasks(self):
+ """
+ An iterator on tasks added to the graph
+ :yields: Iterator over all tasks in the graph
+ """
+ for _, data in self._graph.nodes_iter(data=True):
+ yield data['task']
+
+ def topological_order(self, reverse=False):
+ """
+ Returns topological sort on the graph
+ :param reverse: whether to reverse the sort
+ :return: a list which represents the topological sort
+ """
+ for task_id in topological_sort(self._graph, reverse=reverse):
+ yield self.get_task(task_id)
+
+ def get_dependencies(self, dependent_task):
+ """
+ Iterates over the task's dependencies
+ :param BaseTask dependent_task: The task whose dependencies are requested
+ :yields: Iterator over all tasks which dependency_task depends on
+ :raise: TaskNotInGraphError if dependent_task is not in the graph
+ """
+ if not self.has_tasks(dependent_task):
+ raise TaskNotInGraphError('Task id: {0}'.format(dependent_task.id))
+ for _, dependency_id in self._graph.out_edges_iter(dependent_task.id):
+ yield self.get_task(dependency_id)
+
+ def get_dependents(self, dependency_task):
+ """
+ Iterates over the task's dependents
+ :param BaseTask dependency_task: The task whose dependents are requested
+ :yields: Iterator over all tasks which depend on dependency_task
+ :raise: TaskNotInGraphError if dependency_task is not in the graph
+ """
+ if not self.has_tasks(dependency_task):
+ raise TaskNotInGraphError('Task id: {0}'.format(dependency_task.id))
+ for dependent_id, _ in self._graph.in_edges_iter(dependency_task.id):
+ yield self.get_task(dependent_id)
+
+ # task methods
+
+ def get_task(self, task_id):
+ """
+ Get a task instance that's been inserted to the graph by the task's id
+ :param basestring task_id: The task's id
+ :return: Requested task
+ :rtype: BaseTask
+ :raise: TaskNotInGraphError if no task found in the graph with the given id
+ """
+ if not self._graph.has_node(task_id):
+ raise TaskNotInGraphError('Task id: {0}'.format(task_id))
+ data = self._graph.node[task_id]
+ return data['task']
+
+ @_filter_out_empty_tasks
+ def add_tasks(self, *tasks):
+ """
+ Add a task to the graph
+ :param BaseTask task: The task
+ :return: A list of added tasks
+ :rtype: list
+ """
+ # assert all([isinstance(t, (task.BaseTask, Iterable)) for t in tasks])
+ return_tasks = []
+
+ for t in tasks:
+ if isinstance(t, Iterable):
+ return_tasks += self.add_tasks(*t)
+ elif not self.has_tasks(t):
+ self._graph.add_node(t.id, task=t)
+ return_tasks.append(t)
+
+ return return_tasks
+
+ @_filter_out_empty_tasks
+ def remove_tasks(self, *tasks):
+ """
+ Remove the provided task from the graph
+ :param BaseTask task: The task
+ :return: A list of removed tasks
+ :rtype: list
+ """
+ return_tasks = []
+
+ if len(tasks) > 1:
+ for sub_task in tasks:
+ return_tasks += self.remove_tasks(*sub_task)
+ elif self.has_tasks(tasks[0]):
+ self._graph.remove_node(tasks[0].id)
+ return_tasks.append(tasks[0])
+
+ return return_tasks
+
+ @_filter_out_empty_tasks
+ def has_tasks(self, *tasks):
+ """
+ Check whether a task is in the graph or not
+ :param BaseTask task: The task
+ :return: True if all tasks are in the graph, otherwise True
+ :rtype: list
+ """
+ # assert all(isinstance(t, (task.BaseTask, Iterable)) for t in tasks)
+ return_value = True
+
+ for t in tasks:
+ if isinstance(t, Iterable):
+ return_value &= self.has_tasks(*t)
+ else:
+ return_value &= self._graph.has_node(t.id)
+
+ return return_value
+
+ def add_dependency(self, dependent, dependency):
+ """
+ Add a dependency for one item (task, sequence or parallel) on another
+ The dependent will only be executed after the dependency terminates
+ If either of the items is either a sequence or a parallel,
+ multiple dependencies may be added
+ :param BaseTask|_TasksArrangement dependent: The dependent (task, sequence or parallel)
+ :param BaseTask|_TasksArrangement dependency: The dependency (task, sequence or parallel)
+ :return: True if the dependency between the two hadn't already existed, otherwise False
+ :rtype: bool
+ :raise TaskNotInGraphError if either the dependent or dependency are tasks which
+ are not in the graph
+ """
+ if not (self.has_tasks(dependent) and self.has_tasks(dependency)):
+ raise TaskNotInGraphError()
+
+ if self.has_dependency(dependent, dependency):
+ return False
+
+ if isinstance(dependent, Iterable):
+ for dependent_task in dependent:
+ self.add_dependency(dependent_task, dependency)
+ else:
+ if isinstance(dependency, Iterable):
+ for dependency_task in dependency:
+ self.add_dependency(dependent, dependency_task)
+ else:
+ self._graph.add_edge(dependent.id, dependency.id)
+
+ return True
+
+ def has_dependency(self, dependent, dependency):
+ """
+ Check whether one item (task, sequence or parallel) depends on another
+
+ Note that if either of the items is either a sequence or a parallel,
+ and some of the dependencies exist in the graph but not all of them,
+ this method will return False
+
+ :param BaseTask|_TasksArrangement dependent: The dependent (task, sequence or parallel)
+ :param BaseTask|_TasksArrangement dependency: The dependency (task, sequence or parallel)
+ :return: True if the dependency between the two exists, otherwise False
+ :rtype: bool
+ :raise TaskNotInGraphError if either the dependent or dependency are tasks
+ which are not in the graph
+ """
+ if not (self.has_tasks(dependent) and self.has_tasks(dependency)):
+ raise TaskNotInGraphError()
+
+ return_value = True
+
+ if isinstance(dependent, Iterable):
+ for dependent_task in dependent:
+ return_value &= self.has_dependency(dependent_task, dependency)
+ else:
+ if isinstance(dependency, Iterable):
+ for dependency_task in dependency:
+ return_value &= self.has_dependency(dependent, dependency_task)
+ else:
+ return_value &= self._graph.has_edge(dependent.id, dependency.id)
+
+ return return_value
+
+ def remove_dependency(self, dependent, dependency):
+ """
+ Remove a dependency for one item (task, sequence or parallel) on another
+
+ Note that if either of the items is either a sequence or a parallel, and some of
+ the dependencies exist in the graph but not all of them, this method will not remove
+ any of the dependencies and return False
+
+ :param BaseTask|_TasksArrangement dependent: The dependent (task, sequence or parallel)
+ :param BaseTask|_TasksArrangement dependency: The dependency (task, sequence or parallel)
+ :return: False if the dependency between the two hadn't existed, otherwise True
+ :rtype: bool
+ :raise TaskNotInGraphError if either the dependent or dependency are tasks
+ which are not in the graph
+ """
+ if not (self.has_tasks(dependent) and self.has_tasks(dependency)):
+ raise TaskNotInGraphError()
+
+ if not self.has_dependency(dependent, dependency):
+ return False
+
+ if isinstance(dependent, Iterable):
+ for dependent_task in dependent:
+ self.remove_dependency(dependent_task, dependency)
+ else:
+ if isinstance(dependency, Iterable):
+ for dependency_task in dependency:
+ self.remove_dependency(dependent, dependency_task)
+ else:
+ self._graph.remove_edge(dependent.id, dependency.id)
+
+ return True
+
+ @_filter_out_empty_tasks
+ def sequence(self, *tasks):
+ """
+ Create and insert a sequence into the graph, effectively each task i depends on i-1
+ :param tasks: an iterable of dependencies
+ :return: the provided tasks
+ """
+ if tasks:
+ self.add_tasks(*tasks)
+
+ for i in xrange(1, len(tasks)):
+ self.add_dependency(tasks[i], tasks[i-1])
+
+ return tasks
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/tasks_graph.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/tasks_graph.py b/aria/workflows/api/tasks_graph.py
deleted file mode 100644
index 5160345..0000000
--- a/aria/workflows/api/tasks_graph.py
+++ /dev/null
@@ -1,203 +0,0 @@
-# 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 networkx import DiGraph, topological_sort
-
-from aria.tools.validation import ValidatorMixin
-
-
-class TaskNotFoundError(Exception):
- pass
-
-
-class TaskNotInGraphError(Exception):
- pass
-
-
-class TaskGraph(ValidatorMixin):
- """
- A task graph builder.
- Build a operations flow graph
- """
-
- def __init__(self, name):
- self.name = name
- self.id = str(uuid4())
- self.graph = DiGraph()
-
- def __getattr__(self, attr):
- try:
- return getattr(self.graph, attr)
- except AttributeError:
- return super(TaskGraph, self).__getattribute__(attr)
-
- def __repr__(self):
- return '{name}(id={self.id}, name={self.name}, graph={self.graph!r})'.format(
- name=self.__class__.__name__, self=self)
-
- @property
- def tasks(self):
- """
- An iterator on tasks added to the graph
- """
- for _, data in self.graph.nodes_iter(data=True):
- yield data['task']
-
- @property
- def leaf_tasks(self):
- for task in self.tasks_in_order():
- if not self.graph.predecessors(task.id):
- yield task
-
- def task_tree(self, reverse=False):
- """
- Iterates over the tasks to be executed in topological order and their dependencies.
- :param reverse: reverse the order
- """
- for task in self.tasks_in_order(reverse=reverse):
- yield task, self.task_dependencies(task)
-
- def tasks_in_order(self, reverse=False):
- """
- Iterates over the tasks to be executed in topological order
- :param reverse: reverse the order
- """
- for task_id in topological_sort(self.graph, reverse=reverse):
- yield self.graph.node[task_id]['task']
-
- def has_dependencies(self, task):
- return len(self.task_dependencies(task)) > 0
-
- def task_dependencies(self, task):
- """
- Iterates over the task dependencies
- """
- for task_ids in self.graph.edges_iter(task.id):
- for task_id in task_ids:
- if task.id != task_id:
- yield self.get_task(task_id)
-
- def add_task(self, task):
- """
- Add a task to this graph
- :param WorkflowTask|TaskGraph task: The task
- """
- self.graph.add_node(task.id, task=task)
-
- def get_task(self, task_id):
- """
- Get a task instance that was inserted to this graph by its id
-
- :param basestring task_id: the task id
- :return: requested task
- :rtype: WorkflowTask|TaskGraph
- :raise: TaskNotFoundError if no task found with given id
- """
- try:
- data = self.graph.node[task_id]
- return data['task']
- except KeyError:
- raise TaskNotFoundError('Task id: {0}'.format(task_id))
-
- def remove_task(self, task):
- """
- Remove the provided task from the graph
- :param WorkflowTask|graph task: The task
- """
- self.graph.remove_node(task.id)
-
- def dependency(self, source_task, after):
- """
- Add a dependency between tasks.
- The source task will only be executed after the target task terminates.
- A source task may depend on several tasks,
- In which case it will only be executed after all its target tasks will terminate.
-
- tasks flow order:
- after -> source_task
-
- :param WorkflowTask|TaskGraph source_task: The source task
- :type source_task: WorkflowTask
- :param list after: The target task
- :raise TaskNotInGraphError
- """
- if not self.graph.has_node(source_task.id):
- raise TaskNotInGraphError(
- 'source task {0!r} is not in graph (task id: {0.id})'.format(source_task))
- for target_task in after:
- if not self.graph.has_node(target_task.id):
- raise TaskNotInGraphError(
- 'target task {0!r} is not in graph (task id: {0.id})'.format(target_task))
- self.graph.add_edge(source_task.id, target_task.id)
-
- # workflow creation helper methods
- def chain(self, tasks, after=()):
- """
- create a chain of tasks.
- tasks will be added to the graph with a dependency between
- the tasks by order.
-
- tasks flow order:
- if tasks = (task0, task1, ..., taskn)
- after -> task0 -> task1 -> ... -> taskn
-
- :param tasks: list of WorkflowTask instances.
- :param after: target to the sequence
- """
- for source_task in tasks:
- self.add_task(source_task)
- self.dependency(source_task, after=after)
- after = (source_task,)
-
- def fan_out(self, tasks, after=()):
- """
- create a fan-out.
- tasks will be added to the graph with a dependency to
- the target task.
-
- tasks flow order:
- if tasks = (task0, task1, ..., taskn)
- after -> task0
- |-> task1
- |...
- \-> taskn
-
- :param tasks: list of WorkflowTask instances.
- :param after: target to the tasks
- """
- for source_task in tasks:
- self.add_task(source_task)
- self.dependency(source_task, after=after)
-
- def fan_in(self, source_task, after=None):
- """
- create a fan-in.
- source task will be added to the graph with a dependency to
- the tasks.
-
- tasks flow order:
- if after = (task0, task1, ..., taskn)
- task0\
- task1|-> source_task
- ... |
- taskn/
-
- :param source_task: source to the tasks
- :param after: list of WorkflowTask instances.
- """
- self.add_task(source_task)
- self.dependency(source_task, after=after)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/execute_operation.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/execute_operation.py b/aria/workflows/builtin/execute_operation.py
index 9e87c30..ddbb8e7 100644
--- a/aria/workflows/builtin/execute_operation.py
+++ b/aria/workflows/builtin/execute_operation.py
@@ -24,7 +24,7 @@ from .workflows import execute_operation_on_instance
@workflow
def execute_operation(
- context,
+ ctx,
graph,
operation,
operation_kwargs,
@@ -37,7 +37,7 @@ def execute_operation(
"""
The execute_operation workflow
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext workflow_context: the workflow context
:param TaskGraph graph: the graph which will describe the workflow.
:param basestring operation: the operation name to execute
:param dict operation_kwargs:
@@ -52,7 +52,7 @@ def execute_operation(
subgraphs = {}
# filtering node instances
filtered_node_instances = list(_filter_node_instances(
- context=context,
+ context=ctx,
node_ids=node_ids,
node_instance_ids=node_instance_ids,
type_names=type_names))
@@ -60,31 +60,31 @@ def execute_operation(
if run_by_dependency_order:
filtered_node_instances_ids = set(node_instance.id
for node_instance in filtered_node_instances)
- for node_instance in context.node_instances:
+ for node_instance in ctx.node_instances:
if node_instance.id not in filtered_node_instances_ids:
- subgraphs[node_instance.id] = context.task_graph(
+ subgraphs[node_instance.id] = ctx.task_graph(
name='execute_operation_stub_{0}'.format(node_instance.id))
# registering actual tasks to sequences
for node_instance in filtered_node_instances:
- node_instance_sub_workflow = execute_operation_on_instance(
- context=context,
- graph=graph,
- node_instance=node_instance,
- operation=operation,
- operation_kwargs=operation_kwargs,
- allow_kwargs_override=allow_kwargs_override)
- subgraphs[node_instance.id] = node_instance_sub_workflow
+ graph.add_tasks(
+ execute_operation_on_instance(
+ node_instance=node_instance,
+ operation=operation,
+ operation_kwargs=operation_kwargs,
+ allow_kwargs_override=allow_kwargs_override
+ )
+ )
for _, node_instance_sub_workflow in subgraphs.items():
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
# adding tasks dependencies if required
if run_by_dependency_order:
- for node_instance in context.node_instances:
+ for node_instance in ctx.node_instances:
for relationship_instance in node_instance.relationship_instances:
- graph.dependency(source_task=subgraphs[node_instance.id],
- after=[subgraphs[relationship_instance.target_id]])
+ graph.add_dependency(source_task=subgraphs[node_instance.id],
+ after=[subgraphs[relationship_instance.target_id]])
def _filter_node_instances(context, node_ids=(), node_instance_ids=(), type_names=()):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/heal.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/heal.py b/aria/workflows/builtin/heal.py
index cab2e6e..2f3d538 100644
--- a/aria/workflows/builtin/heal.py
+++ b/aria/workflows/builtin/heal.py
@@ -19,50 +19,49 @@ Builtin heal workflow
from aria import workflow
-from .uninstall import uninstall
-from .install import install
-from .workflows import relationship_tasks
+from .workflows import relationship_tasks, install_node_instance, uninstall_node_instance
+from ..api import task, task_graph
@workflow
-def heal(context, graph, node_instance_id):
+def heal(ctx, graph, node_instance_id):
"""
The heal workflow
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the graph which will describe the workflow.
:param node_instance_id: the id of the node instance to heal
:return:
"""
- failing_node = context.storage.node_instance.get(node_instance_id)
- host_node = context.storage.node_instance.get(failing_node.host_id)
- failed_node_instance_subgraph = _get_contained_subgraph(context, host_node)
+ failing_node = ctx.model.node_instance.get(node_instance_id)
+ host_node = ctx.model.node_instance.get(failing_node.host_id)
+ failed_node_instance_subgraph = _get_contained_subgraph(ctx, host_node)
failed_node_instance_ids = list(n.id for n in failed_node_instance_subgraph)
- targeted_node_instances = [
- context.storage.node_instance.get(relationship_instance.target_id)
- for node_instance in failed_node_instance_subgraph
- for relationship_instance in node_instance.relationship_instances
- if relationship_instance.target_id not in failed_node_instance_ids
- ]
-
- graph.chain([
- heal_uninstall(
- context=context,
- failing_node_instances=failed_node_instance_subgraph,
- targeted_node_instances=targeted_node_instances),
- heal_install(
- context=context,
- failing_node_instances=failed_node_instance_subgraph,
- targeted_node_instances=targeted_node_instances)
- ])
+ targeted_node_instances = [node_instance for node_instance in ctx.node_instances
+ if node_instance.id not in failed_node_instance_ids]
+
+ uninstall_subgraph = task.WorkflowTask(
+ heal_uninstall,
+ failing_node_instances=failed_node_instance_subgraph,
+ targeted_node_instances=targeted_node_instances
+ )
+
+ install_subgraph = task.WorkflowTask(
+ heal_install,
+ failing_node_instances=failed_node_instance_subgraph,
+ targeted_node_instances=targeted_node_instances)
+
+ graph.add_tasks(uninstall_subgraph)
+ graph.add_tasks(install_subgraph)
+ graph.add_dependency(install_subgraph, uninstall_subgraph)
@workflow(suffix_template='{failing_node_instances}')
-def heal_uninstall(context, graph, failing_node_instances, targeted_node_instances):
+def heal_uninstall(ctx, graph, failing_node_instances, targeted_node_instances):
"""
the uninstall part of the heal mechanism
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the task graph to edit.
:param failing_node_instances: the failing nodes to heal.
:param targeted_node_instances: the targets of the relationships where the failing node are
@@ -73,46 +72,53 @@ def heal_uninstall(context, graph, failing_node_instances, targeted_node_instanc
# Create install stub workflow for each unaffected node instance
for node_instance in targeted_node_instances:
- node_instance_sub_workflow = context.task_graph(
+ node_instance_sub_workflow = task_graph.TaskGraph(
name='uninstall_stub_{0}'.format(node_instance.id))
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
- # Create install sub workflow for each failing node
- uninstall(
- context=context,
- graph=graph,
- node_instances=failing_node_instances,
- node_instance_sub_workflows=node_instance_sub_workflows)
+ # Create install sub workflow for each failing node on the same graph
+ node_instance_sub_workflows = node_instance_sub_workflows or {}
- # Add operations for intact nodes depending on a node instance
- # belonging to node_instances
+ # create install sub workflow for every node instance
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = task.WorkflowTask(uninstall_node_instance,
+ node_instance=node_instance)
+ node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
+ graph.add_tasks(node_instance_sub_workflow)
+
+ # create dependencies between the node instance sub workflow
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
+ for relationship_instance in reversed(node_instance.relationship_instances):
+ graph.add_dependency(node_instance_sub_workflows[relationship_instance.target_id],
+ node_instance_sub_workflow)
+
+ # Add operations for intact nodes depending on a node instance belonging to node_instances
for node_instance in targeted_node_instances:
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
for relationship_instance in reversed(node_instance.relationship_instances):
- target_node_instance = context.storage.node_instance.get(
- relationship_instance.target_id)
- if target_node_instance in failing_node_instances:
- after_tasks = [node_instance_sub_workflows[relationship.target_id]
- for relationship in node_instance.relationship_instances]
+ target_node_instance = ctx.model.node_instance.get(relationship_instance.target_id)
+ target_node_instance_subgraph = node_instance_sub_workflows[target_node_instance.id]
+ graph.add_dependency(target_node_instance_subgraph, node_instance_sub_workflow)
- elif target_node_instance in targeted_node_instances:
- after_tasks = [relationship_tasks(
+ if target_node_instance in failing_node_instances:
+ dependency = relationship_tasks(
+ graph=graph,
node_instance=node_instance,
relationship_instance=relationship_instance,
- context=context,
- operation_name='aria.interfaces.relationship_lifecycle.unlink')]
+ context=ctx,
+ operation_name='aria.interfaces.relationship_lifecycle.unlink')
- if after_tasks:
- graph.dependency(source_task=node_instance_sub_workflow, after=after_tasks)
+ graph.add_dependency(node_instance_sub_workflow, dependency)
@workflow(suffix_template='{failing_node_instances}')
-def heal_install(context, graph, failing_node_instances, targeted_node_instances):
+def heal_install(ctx, graph, failing_node_instances, targeted_node_instances):
"""
the install part of the heal mechanism
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the task graph to edit.
:param failing_node_instances: the failing nodes to heal.
:param targeted_node_instances: the targets of the relationships where the failing node are
@@ -123,17 +129,28 @@ def heal_install(context, graph, failing_node_instances, targeted_node_instances
# Create install sub workflow for each unaffected
for node_instance in targeted_node_instances:
- node_instance_sub_workflow = context.task_graph(
+ node_instance_sub_workflow = task_graph.TaskGraph(
name='install_stub_{0}'.format(node_instance.id))
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
+
+ # create install sub workflow for each failing node on the same graph
+ node_instance_sub_workflows = node_instance_sub_workflows or {}
# create install sub workflow for every node instance
- install(
- context=context,
- graph=graph,
- node_instances=failing_node_instances,
- node_instance_sub_workflows=node_instance_sub_workflows)
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = task.WorkflowTask(install_node_instance,
+ node_instance=node_instance)
+ node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
+ graph.add_tasks(node_instance_sub_workflow)
+
+ # create dependencies between the node instance sub workflow
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
+ if node_instance.relationship_instances:
+ dependencies = [node_instance_sub_workflows[relationship_instance.target_id]
+ for relationship_instance in node_instance.relationship_instances]
+ graph.add_dependency(node_instance_sub_workflow, dependencies)
# Add operations for intact nodes depending on a node instance
# belonging to node_instances
@@ -141,35 +158,33 @@ def heal_install(context, graph, failing_node_instances, targeted_node_instances
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
for relationship_instance in node_instance.relationship_instances:
- target_node_instance = context.storage.node_instance.get(
- relationship_instance.target_id)
- if target_node_instance in failing_node_instances:
- after_tasks = [node_instance_sub_workflows[relationship.target_id]
- for relationship in node_instance.relationship_instances]
+ target_node_instance = ctx.model.node_instance.get(relationship_instance.target_id)
+ target_node_instance_subworkflow = node_instance_sub_workflows[target_node_instance.id]
+ graph.add_dependency(node_instance_sub_workflow, target_node_instance_subworkflow)
- elif target_node_instance in targeted_node_instances:
- after_tasks = [relationship_tasks(
+ if target_node_instance in failing_node_instances:
+ dependent = relationship_tasks(
+ graph=graph,
node_instance=node_instance,
relationship_instance=relationship_instance,
- context=context,
- operation_name='aria.interfaces.relationship_lifecycle.establish')]
+ context=ctx,
+ operation_name='aria.interfaces.relationship_lifecycle.establish')
- if after_tasks:
- graph.dependency(source_task=node_instance_sub_workflow, after=after_tasks)
+ graph.add_dependency(dependent, node_instance_sub_workflow)
def _get_contained_subgraph(context, host_node_instance):
- contained_instances = set(node_instance
- for node_instance in context.node_instances
- if node_instance.host_id == host_node_instance.id and
- node_instance.id != node_instance.host_id)
- result = {host_node_instance}
+ contained_instances = [node_instance
+ for node_instance in context.node_instances
+ if node_instance.host_id == host_node_instance.id and
+ node_instance.id != node_instance.host_id]
+ result = [host_node_instance]
if not contained_instances:
return result
- result.update(contained_instances)
+ result.extend(contained_instances)
for node_instance in contained_instances:
- result.update(_get_contained_subgraph(context, node_instance))
+ result.extend(_get_contained_subgraph(context, node_instance))
- return result
+ return set(result)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/install.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/install.py b/aria/workflows/builtin/install.py
index 35d2968..0ab3ad6 100644
--- a/aria/workflows/builtin/install.py
+++ b/aria/workflows/builtin/install.py
@@ -20,13 +20,14 @@ Builtin install workflow
from aria import workflow
from .workflows import install_node_instance
+from ..api import task
@workflow
-def install(context, graph, node_instances=(), node_instance_sub_workflows=None):
+def install(ctx, graph, node_instances=(), node_instance_sub_workflows=None):
"""
The install workflow
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the graph which will describe the workflow.
:param node_instances: the node instances on which to run the workflow
:param dict node_instance_sub_workflows: a dictionary of subworkflows with id as key and
@@ -34,21 +35,19 @@ def install(context, graph, node_instances=(), node_instance_sub_workflows=None)
:return:
"""
node_instance_sub_workflows = node_instance_sub_workflows or {}
- node_instances = node_instances or list(context.node_instances)
+ node_instances = node_instances or list(ctx.node_instances)
# create install sub workflow for every node instance
for node_instance in node_instances:
- node_instance_sub_workflow = install_node_instance(
- context=context,
- node_instance=node_instance)
+ node_instance_sub_workflow = task.WorkflowTask(install_node_instance,
+ node_instance=node_instance)
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
# create dependencies between the node instance sub workflow
for node_instance in node_instances:
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
- graph.dependency(
- source_task=node_instance_sub_workflow,
- after=[
- node_instance_sub_workflows[relationship.target_id]
- for relationship in node_instance.relationship_instances])
+ if node_instance.relationship_instances:
+ dependencies = [node_instance_sub_workflows[relationship_instance.target_id]
+ for relationship_instance in node_instance.relationship_instances]
+ graph.add_dependency(node_instance_sub_workflow, dependencies)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/uninstall.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/uninstall.py b/aria/workflows/builtin/uninstall.py
index 47c8259..f4e965c 100644
--- a/aria/workflows/builtin/uninstall.py
+++ b/aria/workflows/builtin/uninstall.py
@@ -20,10 +20,11 @@ Builtin uninstall workflow
from aria import workflow
from .workflows import uninstall_node_instance
+from ..api import task
@workflow
-def uninstall(context, graph, node_instances=(), node_instance_sub_workflows=None):
+def uninstall(ctx, graph, node_instances=(), node_instance_sub_workflows=None):
"""
The uninstall workflow
:param WorkflowContext context: the workflow context
@@ -34,20 +35,18 @@ def uninstall(context, graph, node_instances=(), node_instance_sub_workflows=Non
:return:
"""
node_instance_sub_workflows = node_instance_sub_workflows or {}
- node_instances = node_instances or list(context.node_instances)
+ node_instances = node_instances or list(ctx.node_instances)
# create install sub workflow for every node instance
for node_instance in node_instances:
- node_instance_sub_workflow = uninstall_node_instance(
- context=context,
- node_instance=node_instance)
+ node_instance_sub_workflow = task.WorkflowTask(uninstall_node_instance,
+ node_instance=node_instance)
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
# create dependencies between the node instance sub workflow
for node_instance in node_instances:
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
for relationship_instance in reversed(node_instance.relationship_instances):
- graph.dependency(
- source_task=node_instance_sub_workflows[relationship_instance.target_id],
- after=[node_instance_sub_workflow])
+ graph.add_dependency(node_instance_sub_workflows[relationship_instance.target_id],
+ node_instance_sub_workflow)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/workflows.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/workflows.py b/aria/workflows/builtin/workflows.py
index b6fbb94..4bff7cd 100644
--- a/aria/workflows/builtin/workflows.py
+++ b/aria/workflows/builtin/workflows.py
@@ -21,6 +21,8 @@ from itertools import groupby
from aria import workflow
+from ..api import task
+
__all__ = (
'install_node_instance',
@@ -32,43 +34,43 @@ __all__ = (
# Install node instance workflow and sub workflows
@workflow(suffix_template='{node_instance.id}')
-def install_node_instance(context, graph, node_instance):
+def install_node_instance(ctx, graph, node_instance):
"""
A workflow which installs a node instance.
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the tasks graph of which to edit
:param node_instance: the node instance to install
:return:
"""
- create_node_instance = context.operation(
+ create_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.create.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.create'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.create'],
node_instance=node_instance
)
- configure_node_instance = context.operation(
+ configure_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.configure.{0}'.format(node_instance.id),
operation_details=node_instance.node.operations['aria.interfaces.lifecycle.configure'],
node_instance=node_instance
)
- start_node_instance = context.operation(
+ start_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.start.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.start'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.start'],
node_instance=node_instance
)
- graph.chain(tasks=[
+
+ graph.sequence(
create_node_instance,
- preconfigure_relationship(context=context, node_instance=node_instance),
+ preconfigure_relationship(graph, ctx, node_instance),
configure_node_instance,
- postconfigure_relationship(context=context, node_instance=node_instance),
+ postconfigure_relationship(graph, ctx, node_instance),
start_node_instance,
- establish_relationship(context=context, node_instance=node_instance),
- ])
+ establish_relationship(graph, ctx, node_instance)
+ )
+ return graph
-@workflow(suffix_template='{node_instance.id}')
-def preconfigure_relationship(context, graph, node_instance):
+
+def preconfigure_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -76,14 +78,14 @@ def preconfigure_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- graph.chain(tasks=relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.preconfigure',
- context=context,
- node_instance=node_instance))
+ context=ctx,
+ node_instance=node_instance)
-@workflow(suffix_template='{node_instance.id}')
-def postconfigure_relationship(context, graph, node_instance):
+def postconfigure_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -91,14 +93,14 @@ def postconfigure_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- graph.chain(tasks=relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.postconfigure',
- context=context,
- node_instance=node_instance))
+ context=ctx,
+ node_instance=node_instance)
-@workflow(suffix_template='{node_instance.id}')
-def establish_relationship(context, graph, node_instance):
+def establish_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -106,16 +108,17 @@ def establish_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- graph.chain(tasks=relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.establish',
- context=context,
- node_instance=node_instance))
+ context=ctx,
+ node_instance=node_instance)
# Uninstall node instance workflow and subworkflows
@workflow(suffix_template='{node_instance.id}')
-def uninstall_node_instance(graph, context, node_instance):
+def uninstall_node_instance(ctx, graph, node_instance):
"""
A workflow which uninstalls a node instance.
:param WorkflowContext context: the workflow context
@@ -123,28 +126,25 @@ def uninstall_node_instance(graph, context, node_instance):
:param node_instance: the node instance to uninstall
:return:
"""
- stop_node_instance = context.operation(
+ stop_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.stop.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.stop'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.stop'],
node_instance=node_instance
)
- delete_node_instance = context.operation(
+ delete_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.delete.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.delete'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.delete'],
node_instance=node_instance
)
- graph.chain(tasks=[
+ graph.sequence(
stop_node_instance,
- unlink_relationship(context=context, node_instance=node_instance),
- delete_node_instance,
- ])
+ unlink_relationship(graph, ctx, node_instance),
+ delete_node_instance
+ )
-@workflow(suffix_template='{node_instance.id}')
-def unlink_relationship(context, graph, node_instance):
+def unlink_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -152,19 +152,15 @@ def unlink_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- tasks = relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.unlink',
- context=context,
+ context=ctx,
node_instance=node_instance
)
- graph.chain(tasks=tasks)
- return tasks
-@workflow(suffix_template='{node_instance.id}.{operation}')
def execute_operation_on_instance(
- context,
- graph,
node_instance,
operation,
operation_kwargs,
@@ -187,16 +183,18 @@ def execute_operation_on_instance(
node_instance=node_instance,
operation_name=operation)
- graph.add_task(
- context.operation(
- name=task_name,
- operation_details=node_instance.node.operations[operation],
- node_instance=node_instance,
- parameters=operation_kwargs)
- )
+ return task.OperationTask(
+ name=task_name,
+ operation_details=node_instance.node.operations[operation],
+ node_instance=node_instance,
+ inputs=operation_kwargs)
-def relationships_tasks(operation_name, context, node_instance):
+
+def relationships_tasks(graph,
+ operation_name,
+ context,
+ node_instance):
"""
Creates a relationship task (source and target) for all of a node_instance relationships.
:param basestring operation_name: the relationship operation name.
@@ -211,17 +209,24 @@ def relationships_tasks(operation_name, context, node_instance):
sub_tasks = []
for index, (_, relationship_group) in enumerate(relationships_groups):
for relationship_instance in relationship_group:
- relationship_subgraph = relationship_tasks(
+ relationship_operations = relationship_tasks(
+ graph=graph,
node_instance=node_instance,
relationship_instance=relationship_instance,
context=context,
operation_name=operation_name,
index=index)
- sub_tasks.append(relationship_subgraph)
- return sub_tasks
+ sub_tasks.append(relationship_operations)
+
+ return graph.sequence(*sub_tasks) if sub_tasks else sub_tasks
-def relationship_tasks(node_instance, relationship_instance, context, operation_name, index=None):
+def relationship_tasks(graph,
+ node_instance,
+ relationship_instance,
+ context,
+ operation_name,
+ index=None):
"""
Creates a relationship task source and target.
:param NodeInstance node_instance: the node instance of the relationship
@@ -232,29 +237,20 @@ def relationship_tasks(node_instance, relationship_instance, context, operation_
:return:
"""
index = index or node_instance.relationship_instances.index(relationship_instance)
- sub_workflow_name = '{name}.{index}.{node_instance.id}'.format(
- name=operation_name,
- index=index,
- node_instance=node_instance,
- )
operation_name_template = '{name}.{index}.{{0}}.<{source_id}, {target_id}>'.format(
name=operation_name,
index=index,
source_id=node_instance.id,
target_id=relationship_instance.target_id,
)
- source_operation = context.operation(
+ source_operation = task.OperationTask(
name=operation_name_template.format('source'),
node_instance=node_instance,
operation_details=relationship_instance.relationship.source_operations[
operation_name])
- target_operation = context.operation(
+ target_operation = task.OperationTask(
name=operation_name_template.format('target'),
- node_instance=context.storage.node_instance.get(
- relationship_instance.target_id),
+ node_instance=context.model.node_instance.get(relationship_instance.target_id),
operation_details=relationship_instance.relationship.target_operations[
operation_name])
- sub_graph = context.task_graph(name=sub_workflow_name)
- sub_graph.add_task(source_operation)
- sub_graph.add_task(target_operation)
- return sub_graph
+ return graph.add_tasks(source_operation, target_operation)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/core/__init__.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/__init__.py b/aria/workflows/core/__init__.py
index ae1e83e..08ccff7 100644
--- a/aria/workflows/core/__init__.py
+++ b/aria/workflows/core/__init__.py
@@ -12,3 +12,5 @@
# 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 . import translation, task
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/core/executor.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/executor.py b/aria/workflows/core/executor.py
index ace445a..0a5cd67 100644
--- a/aria/workflows/core/executor.py
+++ b/aria/workflows/core/executor.py
@@ -134,8 +134,7 @@ class MultiprocessExecutor(BaseExecutor):
self._listener_thread = threading.Thread(target=self._listener)
self._listener_thread.daemon = True
self._listener_thread.start()
- self._pool = multiprocessing.Pool(processes=pool_size,
- maxtasksperchild=1)
+ self._pool = multiprocessing.Pool(processes=pool_size)
def execute(self, task):
self._tasks[task.id] = task