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/10/20 15:15:09 UTC
incubator-ariatosca git commit: pylint-aria-workflow-builtin
Repository: incubator-ariatosca
Updated Branches:
refs/heads/wf-wip b57fab5d4 -> 606f2364a
pylint-aria-workflow-builtin
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/606f2364
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/606f2364
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/606f2364
Branch: refs/heads/wf-wip
Commit: 606f2364a062493076c093d3ac6f1e0686aea22c
Parents: b57fab5
Author: mxmrlv <mx...@gmail.com>
Authored: Wed Oct 19 20:01:32 2016 +0300
Committer: mxmrlv <mx...@gmail.com>
Committed: Thu Oct 20 18:14:58 2016 +0300
----------------------------------------------------------------------
aria/.pylintrc | 6 +-
aria/workflows/builtin/__init__.py | 11 +-
.../builtin/deployment_modification.py | 221 ----------
aria/workflows/builtin/execute_operation.py | 34 +-
aria/workflows/builtin/heal.py | 34 +-
aria/workflows/builtin/install.py | 14 +-
aria/workflows/builtin/scale.py | 416 -------------------
aria/workflows/builtin/uninstall.py | 14 +-
aria/workflows/builtin/update.py | 21 -
aria/workflows/builtin/workflows.py | 108 ++++-
10 files changed, 183 insertions(+), 696 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/.pylintrc
----------------------------------------------------------------------
diff --git a/aria/.pylintrc b/aria/.pylintrc
index a70a41a..eb188a0 100644
--- a/aria/.pylintrc
+++ b/aria/.pylintrc
@@ -332,14 +332,14 @@ exclude-protected=_asdict,_fields,_replace,_source,_make
[DESIGN]
# Maximum number of arguments for function / method
-max-args=15
+max-args=20
# Argument names that match this expression will be ignored. Default to name
# with leading underscore
ignored-argument-names=_.*
# Maximum number of locals for function / method body
-max-locals=15
+max-locals=20
# Maximum number of return / yield for function / method body
max-returns=6
@@ -354,7 +354,7 @@ max-statements=50
max-parents=7
# Maximum number of attributes for a class (see R0902).
-max-attributes=15
+max-attributes=20
# Minimum number of public methods for a class (see R0903).
min-public-methods=0
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/__init__.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/__init__.py b/aria/workflows/builtin/__init__.py
index c868fe7..0449a8e 100644
--- a/aria/workflows/builtin/__init__.py
+++ b/aria/workflows/builtin/__init__.py
@@ -13,20 +13,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+"""
+A set of builtin workflows
+"""
+
from .install import install
from .uninstall import uninstall
from .execute_operation import execute_operation
from .heal import heal
-from .scale import scale_entity, scale
-from .update import update
__all__ = [
'install',
- 'uninstall'
+ 'uninstall',
'execute_operation',
'heal',
- 'scale'
- 'scale_entity',
- 'update'
]
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/deployment_modification.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/deployment_modification.py b/aria/workflows/builtin/deployment_modification.py
deleted file mode 100644
index b4300b1..0000000
--- a/aria/workflows/builtin/deployment_modification.py
+++ /dev/null
@@ -1,221 +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.
-
-import itertools
-from datetime import datetime
-
-from aria.storage import models
-from aria.deployment import modify_deployment as extract_deployment_modification
-
-
-def modify_deployment(context, modified_nodes, modification_context=None):
- deployment = context.deployment
- active_modifications = filter(
- lambda dm: dm == models.DeploymentModification.STARTED,
- context.storage.deployment_modification)
- if active_modifications:
- # TODO: raise proper exception
- raise Exception(
- 'Cannot start deployment modification while there are '
- 'existing started deployment modifications. Currently '
- 'started deployment modifications: {0}'
- .format(active_modifications))
-
- nodes = set(context.nodes)
- node_instances = set(context.node_instances)
- node_instances_modification = extract_deployment_modification(
- nodes=nodes,
- previous_nodes=nodes,
- previous_node_instances=node_instances,
- scaling_groups=deployment.scaling_groups,
- modified_nodes=modified_nodes)
-
- modification = models.DeploymentModification(
- created_at=datetime.now(),
- ended_at=None,
- status=models.DeploymentModification.STARTED,
- deployment_id=context.deployment_id,
- modified_nodes=modified_nodes,
- added_and_related=node_instances_modification['added_and_related'],
- removed_and_related=node_instances_modification['removed_and_related'],
- extended_and_related=node_instances_modification['extended_and_related'],
- reduced_and_related=node_instances_modification['reduced_and_related'],
- # before_modification=node_instances,
- # node_instances=node_instances_modification,
- context=modification_context)
-
- context.storage.deployment_modification.store(modification)
-
- for node_id, modified_node in modified_nodes.items():
- if node_id in deployment.scaling_groups:
- deployment.scaling_groups[node_id]['properties'].update({
- 'planned_instances': modified_node['instances']
- })
- else:
- node = context.storage.node.get(node_id)
- node.planned_number_of_instances = modified_node['instances']
- context.storage.node.store(node)
- context.storage.deployment.store(deployment)
-
- added_and_related = node_instances_modification['added_and_related']
- added_node_instances = []
- for node_instance in added_and_related:
- if hasattr(node_instance, 'modification') and node_instance.modification == 'added':
- added_node_instances.append(node_instance)
- else:
- current_instance = context.storage.node_instance.get(node_instance.id)
- target_names = [r.target_id for r in current_instance.node.relationships]
- current_relationship_groups = {
- target_name: list(group)
- for target_name, group in itertools.groupby(
- current_instance.relationship_instances,
- key=lambda r: r.target_name)
- }
- new_relationship_groups = {
- target_name: list(group)
- for target_name, group in itertools.groupby(
- node_instance.relationship_instances,
- key=lambda r: r.target_name)
- }
- new_relationships = []
- for target_name in target_names:
- new_relationships += current_relationship_groups.get(
- target_name, [])
- new_relationships += new_relationship_groups.get(
- target_name, [])
-
- updated_node_instance = models.NodeInstance(
- id=node_instance.id,
- deployment_id=context.deployment_id,
- relationship_instances=new_relationships,
- version=current_instance.version,
- node=context.storage.node.get(node_instance.node.id),
- host_id=None,
- runtime_properties={})
-
- context.storage.node_instance.store(updated_node_instance)
-
- _create_deployment_node_instances(context, added_node_instances)
- return modification
-
-
-def _create_deployment_node_instances(context,
- dsl_node_instances):
- node_instances = []
- for node_instance in dsl_node_instances:
- instance = models.NodeInstance(
- id=node_instance.id,
- node=node_instance.node,
- host_id=node_instance.host_id,
- relationship_instances=node_instance.relationship_instances,
- deployment_id=context.deployment_id,
- runtime_properties={},
- version=None,
- scaling_groups=node_instance.scaling_groups)
- node_instances.append(instance)
-
- for node_instance in node_instances:
- context.storage.node_instance.store(node_instance)
-
-
-def finish_deployment_modification(context, modification_id):
- modification = context.storage.deployment_modification.get(modification_id)
-
- if modification.status in models.DeploymentModification.END_STATES:
- raise Exception(
- 'Cannot finish deployment modification: {0}. It is already in'
- ' {1} status.'.format(modification_id,
- modification.status))
-
- for node_id, modified_node in modification.modified_nodes.items():
- if node_id in context.deployment.scaling_groups:
- context.deployment.scaling_groups[node_id]['properties'].update({
- 'current_instances': modified_node['instances']
- })
- else:
- node_dict = context.storage.node.get(node_id).fields_dict
- node_dict['number_of_instances'] = modified_node['instances']
- context.storage.node.store(models.Node(**node_dict))
-
- context.storage.deployment.store(context.deployment)
-
- for node_instance in modification.removed_and_related:
- if node_instance.get('modification') == 'removed':
- context.storage.node_instance.delete(node_instance.id)
- pass
- else:
- removed_relationship_target_ids = set(
- [rel.target_id for rel in node_instance.relationships])
- current = context.storage.node_instance.get(node_instance.id)
- new_relationships = [rel for rel in current.relationships
- if rel.target_id not in removed_relationship_target_ids]
- context.storage.node_instance.store(models.NodeInstance(
- id=node_instance.id,
- relationships=new_relationships,
- version=current.version,
- node_id=None,
- host_id=None,
- deployment_id=None,
- state=None,
- runtime_properties=None))
-
- now = datetime.now()
-
- context.storage.deployment_modification.store(
- models.DeploymentModification(
- id=modification_id,
- status=models.DeploymentModification.FINISHED,
- ended_at=now,
- created_at=modification.created_at,
- deployment_id=modification.deployment_id,
- modified_nodes=modification.modified_nodes,
- added_and_related=modification.added_and_related,
- removed_and_related=modification.removed_and_related,
- extended_and_related=modification.extended_and_related,
- reduced_and_related=modification.reduced_and_related,
- context=None))
-
-
-def rollback_deployment_modification(context, modification_id):
- pass
- # modification = context.storage.deployment_modification.get(modification_id)
- #
- # if modification.status in models.DeploymentModification.END_STATES:
- # raise Exception(
- # 'Cannot rollback deployment modification: {0}. It is already '
- # 'in {1} status.'.format(modification_id,
- # modification.status))
- #
- # # before_rollback_node_instnaces = [instance.fields_dict for instance in context.node_instances]
- # for instance in context.node_instances:
- # context.storage.node_instance.delete(instance.id)
- #
- # for instance in modification.before_modification:
- # context.storage.node_instance.store(models.NodeInstance(**instance))
- # nodes_num_instances = dict((node.id, node) for node in context.nodes)
- #
- # modified_nodes = modification.modified_nodes
- # for node_id, modified_node in modified_nodes.items():
- # if node_id in context.deployment.scaling_groups:
- # props = context.deployment.scaling_groups[node_id]['properties']
- # props['planned_instances'] = props['current_instances']
- # else:
- # rolled_back_node_dict = context.storage.node.get(node_id).fields_dict
- # rolled_back_node_dict['planned_number_of_instances'] = nodes_num_instances[node_id].number_of_instances
- # context.storage.node.store(models.Node(**rolled_back_node_dict))
- #
- # context.storage.deployment.store(context.deployment)
- #
- # return context.storage.deployment_modification.get(modification_id)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/execute_operation.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/execute_operation.py b/aria/workflows/builtin/execute_operation.py
index 6a5cbd4..80a598d 100644
--- a/aria/workflows/builtin/execute_operation.py
+++ b/aria/workflows/builtin/execute_operation.py
@@ -13,6 +13,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+"""
+Builtin execute_operation workflow
+"""
+
from aria import workflow
from .workflows import execute_operation_on_instance
@@ -30,6 +34,21 @@ def execute_operation(
node_ids,
node_instance_ids,
**kwargs):
+ """
+ The execute_operation workflow
+
+ :param WorkflowContext 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:
+ :param bool allow_kwargs_override:
+ :param bool run_by_dependency_order:
+ :param type_names:
+ :param node_ids:
+ :param node_instance_ids:
+ :param kwargs:
+ :return:
+ """
subgraphs = {}
# filtering node instances
filtered_node_instances = list(_filter_node_instances(
@@ -70,8 +89,17 @@ def execute_operation(
def _filter_node_instances(context, node_ids=(), node_instance_ids=(), type_names=()):
+ def _is_node_by_id(node_id):
+ return not node_ids or node_id in node_ids
+
+ def _is_node_instance_by_id(node_instance_id):
+ return not node_instance_ids or node_instance_id in node_instance_ids
+
+ def _is_node_by_type(node_type_hierarchy):
+ return not type_names or node_type_hierarchy in type_names
+
for node_instance in context.node_instances:
- if ((not node_instance_ids or node_instance.id in node_instance_ids) and
- (not node_ids or node_instance.node.id in node_ids) and
- (not type_names or node_instance.node.type_hierarchy in type_names)):
+ if all((_is_node_by_id(node_instance.node.id),
+ _is_node_instance_by_id(node_instance.id),
+ _is_node_by_type(node_instance.node.type_hierarchy))):
yield node_instance
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/heal.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/heal.py b/aria/workflows/builtin/heal.py
index 137f1e4..cab2e6e 100644
--- a/aria/workflows/builtin/heal.py
+++ b/aria/workflows/builtin/heal.py
@@ -13,6 +13,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+"""
+Builtin heal workflow
+"""
+
from aria import workflow
from .uninstall import uninstall
@@ -22,6 +26,14 @@ from .workflows import relationship_tasks
@workflow
def heal(context, graph, node_instance_id):
+ """
+ The heal workflow
+
+ :param WorkflowContext context: 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)
@@ -48,6 +60,15 @@ def heal(context, graph, node_instance_id):
@workflow(suffix_template='{failing_node_instances}')
def heal_uninstall(context, graph, failing_node_instances, targeted_node_instances):
+ """
+ the uninstall part of the heal mechanism
+ :param WorkflowContext context: 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
+ source
+ :return:
+ """
node_instance_sub_workflows = {}
# Create install stub workflow for each unaffected node instance
@@ -89,6 +110,15 @@ def heal_uninstall(context, graph, failing_node_instances, targeted_node_instanc
@workflow(suffix_template='{failing_node_instances}')
def heal_install(context, graph, failing_node_instances, targeted_node_instances):
+ """
+ the install part of the heal mechanism
+ :param WorkflowContext context: 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
+ source
+ :return:
+ """
node_instance_sub_workflows = {}
# Create install sub workflow for each unaffected
@@ -123,7 +153,7 @@ def heal_install(context, graph, failing_node_instances, targeted_node_instances
relationship_instance=relationship_instance,
context=context,
operation_name='aria.interfaces.relationship_lifecycle.establish')]
-
+
if after_tasks:
graph.dependency(source_task=node_instance_sub_workflow, after=after_tasks)
@@ -143,5 +173,3 @@ def _get_contained_subgraph(context, host_node_instance):
result.update(_get_contained_subgraph(context, node_instance))
return result
-
-
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/install.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/install.py b/aria/workflows/builtin/install.py
index 15e6516..35d2968 100644
--- a/aria/workflows/builtin/install.py
+++ b/aria/workflows/builtin/install.py
@@ -13,6 +13,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+"""
+Builtin install workflow
+"""
+
from aria import workflow
from .workflows import install_node_instance
@@ -20,7 +24,15 @@ from .workflows import install_node_instance
@workflow
def install(context, graph, node_instances=(), node_instance_sub_workflows=None):
-
+ """
+ The install workflow
+ :param WorkflowContext context: 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
+ TaskGraph (or OperationContext) as value
+ :return:
+ """
node_instance_sub_workflows = node_instance_sub_workflows or {}
node_instances = node_instances or list(context.node_instances)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/scale.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/scale.py b/aria/workflows/builtin/scale.py
deleted file mode 100644
index f39a1a8..0000000
--- a/aria/workflows/builtin/scale.py
+++ /dev/null
@@ -1,416 +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 aria import workflow
-from aria.workflows.core.engine import Engine
-from .install import install
-from .uninstall import uninstall
-
-from .deployment_modification import modify_deployment, finish_deployment_modification, \
- rollback_deployment_modification
-
-
-def scale(context, node_id, delta, scale_compute):
- return scale_entity(
- context=context,
- scalable_entity_name=node_id,
- delta=delta,
- scale_compute=scale_compute)
-
-
-# TODO: 1. the screening of which nodes were added doesn't work
-
-@workflow(simple_workflow=False)
-def scale_entity(context, graph, scalable_entity_name, delta, scale_compute):
- engine = Engine(context.concurrency_count)
- if isinstance(delta, basestring):
- try:
- delta = int(delta)
- except ValueError:
- raise ValueError('The delta parameter must be a number. Got: {0}'.format(delta))
-
- if delta == 0:
- context.logger.info('delta parameter is 0, so no scaling will take place.')
- return
-
- scaling_group = context.deployment.scaling_groups.get(scalable_entity_name)
- if scaling_group:
- curr_num_instances = scaling_group['properties']['current_instances']
- planned_num_instances = curr_num_instances + delta
- scale_id = scalable_entity_name
- else:
- node = context.storage.node.get(scalable_entity_name)
- if not node:
- raise ValueError("No scalable entity named {0} was found".format(
- scalable_entity_name))
- host_node = context.storage.node.get(node.host_id)
- scaled_node = host_node if (scale_compute and host_node) else node
- curr_num_instances = scaled_node.number_of_instances
- planned_num_instances = curr_num_instances + delta
- scale_id = scaled_node.id
-
- if planned_num_instances < 0:
- raise ValueError('Provided delta: {0} is illegal. current number of '
- 'instances of entity {1} is {2}'
- .format(delta,
- scalable_entity_name,
- curr_num_instances))
-
- modification = modify_deployment(
- context,
- {
- scale_id: {
- 'instances': planned_num_instances
-
- # These following parameters are not exposed at the moment,
- # but should be used to control which node instances get scaled in
- # (when scaling in).
- # They are mentioned here, because currently, the modification API
- # is not very documented.
- # Special care should be taken because if `scale_compute == True`
- # (which is the default), then these ids should be the compute node
- # instance ids which are not necessarily instances of the node
- # specified by `scalable_entity_name`.
-
- # Node instances denoted by these instance ids should be *kept* if
- # possible.
- # 'removed_ids_exclude_hint': [],
-
- # Node instances denoted by these instance ids should be *removed*
- # if possible.
- # 'removed_ids_include_hint': []
- }
- }
- )
- try:
- context.logger.info('Deployment modification started. '
- '[modification_id={0}]'.format(modification.id))
- if delta > 0:
- added, related = [], []
- for node_instance in modification.added_and_related:
- if hasattr(node_instance, 'modification') and node_instance.modification == 'added':
- added.append(node_instance)
- else:
- related.append(node_instance)
- try:
- graph.add_task(_scale_install(
- graph=graph,
- context=context,
- scaling_up_node_instances=added,
- unaffected_node_instances=related))
- except:
- context.logger.error('Scale out failed, scaling back in.')
- for task in graph.tasks:
- graph.remove_task(task)
- graph.add_task(_scale_uninstall(
- graph=graph,
- context=context,
- scaling_down_node_instances=added,
- unaffected_node_instances=related))
- raise
- else:
- removed, related = [], []
- for node_instance in modification.removed_and_related:
- if hasattr(node_instance, 'modifictation') and node_instance.pop('modification') == 'removed':
- removed.append(node_instance)
- else:
- related.append(node_instance)
- graph.add_task(_scale_uninstall(
- context=context,
- scaling_down_node_instances=removed,
- related_nodes=related))
- except:
- context.logger.warn('Rolling back deployment modification. '
- '[modification_id={0}]'.format(modification.id))
- try:
- rollback_deployment_modification(context, modification.id)
- except:
- context.logger.warn('Deployment modification rollback failed. The '
- 'deployment model is most likely in some corrupted'
- ' state.'
- '[modification_id={0}]'.format(modification.id))
- raise
- raise
- else:
- engine.create_workflow(context, graph)
- engine.execute()
- try:
- finish_deployment_modification(context, modification.id)
- except:
- context.logger.warn('Deployment modification finish failed. The '
- 'deployment model is most likely in some corrupted'
- ' state.'
- '[modification_id={0}]'.format(modification.id))
- raise
-
-
-@workflow(simple_workflow=False)
-def _scale_uninstall(context, graph, scaling_down_node_instances, unaffected_node_instances):
- node_instance_sub_workflows = {}
-
- # Create install sub workflow for each unaffected
- for node_instance in unaffected_node_instances:
- node_instance_sub_workflow = uninstall_stub_subworkflow(
- sub_workflow_name='uninstall_stub_{0}'.format(node_instance.id),
- graph=graph,
- context=context)
- node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
-
- # Create install sub workflow for each failing node
- uninstall(
- context=context,
- graph=graph,
- node_instances=scaling_down_node_instances,
- node_instance_sub_workflows=node_instance_sub_workflows)
-
- # Add operations for intact nodes depending on a node instance
- # belonging to node_instances
- intact_node_relationship_operation = 'aria.interfaces.relationship_lifecycle.unlink'
- for node_instance in unaffected_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)
- after_tasks = []
- if target_node_instance in scaling_down_node_instances:
- after_tasks.extend(
- [node_instance_sub_workflows[relationship.target_id]
- for relationship in node_instance.relationship_instances])
-
- elif target_node_instance in unaffected_node_instances:
- intact_tasks = relationship_tasks_subworkflow(
- graph=graph,
- context=context,
- sub_workflow_name='{0}.{1}'.format(
- intact_node_relationship_operation,
- node_instance.id),
- operation_name=intact_node_relationship_operation,
- node_instance=node_instance,
- relationship_instance=relationship_instance)
- after_tasks.extend(intact_tasks)
-
- graph.dependency(source_task=node_instance_sub_workflow, after=after_tasks)
-
-
-@workflow(simple_workflow=False)
-def _scale_install(context, graph, scaling_up_node_instances, unaffected_node_instances):
- node_instance_sub_workflows = {}
-
- # Create install sub workflow for each unaffected
- for node_instance in unaffected_node_instances:
- node_instance_sub_workflow = install_stub_subworkflow(
- sub_workflow_name='install_stub_{0}'.format(node_instance.id),
- context=context,
- graph=graph)
- node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
-
- # create install sub workflow for every node instance
- install(
- context=context,
- graph=graph,
- node_instances=scaling_up_node_instances,
- node_instance_sub_workflows=node_instance_sub_workflows)
-
- # Add operations for intact nodes depending on a node instance
- # belonging to node_instances
- intact_node_relationship_operation = 'aria.interfaces.relationship_lifecycle.establish'
- for node_instance in unaffected_node_instances:
- node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
-
- for relationship_instance in node_instance.relationship_instances:
- after_tasks = [
- node_instance_sub_workflows[relationship_instance.target_id]
- for relationship_instance in node_instance.relationship_instances]
-
- if context.storage.node_instance.get(relationship_instance.target_id) in unaffected_node_instances:
- intact_tasks = relationship_tasks_subworkflow(
- graph=graph,
- context=context,
- sub_workflow_name='{0}.{1}'.format(
- intact_node_relationship_operation,
- node_instance.id),
- operation_name=intact_node_relationship_operation,
- node_instance=node_instance,
- relationship_instance=relationship_instance)
- after_tasks.extend(intact_tasks)
-
- graph.dependency(source_task=node_instance_sub_workflow, after=after_tasks)
-
-
-@workflow(simple_workflow=False)
-def install_stub_subworkflow(**kwargs):
- pass
-
-
-@workflow(simple_workflow=False)
-def uninstall_stub_subworkflow(**kwargs):
- pass
-
-
-@workflow(simple_workflow=False)
-def install_node_instance_sub_workflow(graph, context, node_instance):
- graph.chain(tasks=[
- context.operation(
- name='aria.interfaces.lifecycle.create.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.create'],
- node_instance=node_instance),
- preconfigure_relationship_sub_workflow(
- sub_workflow_name='preconfigure_{0}'.format(node_instance.id),
- context=context,
- graph=graph,
- node_instance=node_instance),
- context.operation(
- name='aria.interfaces.lifecycle.configure.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.configure'],
- node_instance=node_instance),
- postconfigure_relationship_sub_workflow(
- sub_workflow_name='postconfigure_{0}'.format(node_instance.id),
- context=context,
- graph=graph,
- node_instance=node_instance),
- context.operation(
- name='aria.interfaces.lifecycle.start.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.start'],
- node_instance=node_instance),
- establish_relationship_sub_workflow(
- sub_workflow_name='establish_{0}'.format(node_instance.id),
- context=context,
- graph=graph,
- node_instance=node_instance),
- ])
-
-
-@workflow(simple_workflow=False)
-def uninstall_node_instance_sub_workflow(graph, context, node_instance):
- graph.chain(tasks=[
- # instance.set_state('stopping'),
- # instance.send_event('Stopping node'),
- context.operation(
- name='aria.interfaces.monitoring.stop.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations['aria.interfaces.monitoring.stop'],
- node_instance=node_instance
- ),
- context.operation(
- name='aria.interfaces.lifecycle.stop.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations['aria.interfaces.lifecycle.stop'],
- node_instance=node_instance
- ),
- # instance.set_state('stopped'),
- unlink_relationship_sub_workflow(
- sub_workflow_name='unlink{0}'.format(node_instance.id),
- context=context,
- graph=graph,
- node_instance=node_instance),
- # instance.set_state('deleting'),
- # instance.send_event('Deleting node'),
- context.operation(
- name='aria.interfaces.lifecycle.delete.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations['aria.interfaces.lifecycle.delete'],
- node_instance=node_instance
- )
- # instance.set_state('deleted')
-
- ])
-
-
-@workflow(simple_workflow=False)
-def preconfigure_relationship_sub_workflow(context, graph, node_instance):
- operation_name = 'aria.interfaces.relationship_lifecycle.preconfigure'
- relationship_tasks = []
- for relationship_instance in node_instance.relationship_instances:
- relationship_tasks.append(
- relationship_tasks_subworkflow(
- graph=graph,
- context=context,
- sub_workflow_name='{0}.{1}'.format(operation_name, node_instance.id),
- operation_name=operation_name,
- node_instance=node_instance,
- relationship_instance=relationship_instance)
- )
- graph.chain(tasks=relationship_tasks)
-
-
-@workflow(simple_workflow=False)
-def postconfigure_relationship_sub_workflow(context, graph, node_instance):
- operation_name = 'aria.interfaces.relationship_lifecycle.postconfigure'
- relationship_tasks = []
- for relationship_instance in node_instance.relationship_instances:
- relationship_tasks.append(
- relationship_tasks_subworkflow(
- graph=graph,
- context=context,
- sub_workflow_name='{0}.{1}'.format(operation_name, node_instance.id),
- operation_name=operation_name,
- node_instance=node_instance,
- relationship_instance=relationship_instance)
- )
- graph.chain(tasks=relationship_tasks)
-
-
-@workflow(simple_workflow=False)
-def establish_relationship_sub_workflow(context, graph, node_instance):
- operation_name = 'aria.interfaces.relationship_lifecycle.establish'
- relationship_tasks = []
- for relationship_instance in node_instance.relationship_instances:
- relationship_tasks.append(
- relationship_tasks_subworkflow(
- graph=graph,
- context=context,
- sub_workflow_name='{0}.{1}'.format(operation_name, node_instance.id),
- operation_name=operation_name,
- node_instance=node_instance,
- relationship_instance=relationship_instance)
- )
- graph.chain(tasks=relationship_tasks)
-
-
-@workflow(simple_workflow=False)
-def unlink_relationship_sub_workflow(context, graph, node_instance):
- operation_name = 'aria.interfaces.relationship_lifecycle.unlink'
- relationship_tasks = []
- for relationship_instance in node_instance.relationship_instances:
- relationship_tasks.append(
- relationship_tasks_subworkflow(
- graph=graph,
- context=context,
- sub_workflow_name='{0}.{1}'.format(operation_name, node_instance.id),
- operation_name=operation_name,
- node_instance=node_instance,
- relationship_instance=relationship_instance)
- )
- graph.chain(tasks=relationship_tasks)
-
-
-@workflow(simple_workflow=False)
-def relationship_tasks_subworkflow(graph, context, operation_name, node_instance, relationship_instance):
- source_operation = relationship_instance.relationship.source_operations[operation_name]
- target_instance = context.storage.node_instance.get(relationship_instance.target_id)
- target_operation = relationship_instance.relationship.target_operations[operation_name]
- graph.fan_out([
- context.operation(
- name='source_{0}'.format(operation_name),
- node_instance=node_instance,
- operation_details=source_operation),
- context.operation(
- name='target_{0}'.format(operation_name),
- node_instance=target_instance,
- operation_details=target_operation),
- ])
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/uninstall.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/uninstall.py b/aria/workflows/builtin/uninstall.py
index 85f9bfb..47c8259 100644
--- a/aria/workflows/builtin/uninstall.py
+++ b/aria/workflows/builtin/uninstall.py
@@ -13,6 +13,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+"""
+Builtin uninstall workflow
+"""
+
from aria import workflow
from .workflows import uninstall_node_instance
@@ -20,7 +24,15 @@ from .workflows import uninstall_node_instance
@workflow
def uninstall(context, graph, node_instances=(), node_instance_sub_workflows=None):
-
+ """
+ The uninstall workflow
+ :param WorkflowContext context: 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
+ TaskGraph (or OperationContext) as value
+ :return:
+ """
node_instance_sub_workflows = node_instance_sub_workflows or {}
node_instances = node_instances or list(context.node_instances)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/update.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/update.py b/aria/workflows/builtin/update.py
deleted file mode 100644
index eb7a097..0000000
--- a/aria/workflows/builtin/update.py
+++ /dev/null
@@ -1,21 +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 aria import workflow
-
-
-@workflow
-def update(context, graph):
- pass
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/606f2364/aria/workflows/builtin/workflows.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/workflows.py b/aria/workflows/builtin/workflows.py
index ad4155b..a87dd74 100644
--- a/aria/workflows/builtin/workflows.py
+++ b/aria/workflows/builtin/workflows.py
@@ -13,28 +13,33 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+"""
+A set of builtin workflows.
+"""
+
from itertools import groupby
from aria import workflow
-__all__ = [
- 'create_node_instance',
- 'preconfigure_relationship',
- 'configure_node_instance',
- 'postconfigure_relationship',
- 'start_node_instance',
- 'establish_relationship',
- 'stop_node_instance',
- 'unlink_relationship',
- 'delete_node_instance'
-]
+__all__ = (
+ 'install_node_instance',
+ 'uninstall_node_instance',
+ 'execute_operation_on_instance',
+)
# Install node instance workflow and sub workflows
@workflow(suffix_template='{node_instance.id}')
def install_node_instance(context, graph, node_instance):
+ """
+ A workflow which installs a node instance.
+ :param WorkflowContext context: 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(
name='aria.interfaces.lifecycle.create.{0}'.format(node_instance.id),
operation_details=node_instance.node.operations[
@@ -42,10 +47,9 @@ def install_node_instance(context, graph, node_instance):
node_instance=node_instance
)
configure_node_instance = context.operation(
- name='aria.interfaces.lifecycle.configure.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.configure'],
- node_instance=node_instance
+ 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(
name='aria.interfaces.lifecycle.start.{0}'.format(node_instance.id),
@@ -65,6 +69,13 @@ def install_node_instance(context, graph, node_instance):
@workflow(suffix_template='{node_instance.id}')
def preconfigure_relationship(context, graph, node_instance):
+ """
+
+ :param context:
+ :param graph:
+ :param node_instance:
+ :return:
+ """
graph.chain(tasks=relationships_tasks(
operation_name='aria.interfaces.relationship_lifecycle.preconfigure',
context=context,
@@ -73,6 +84,13 @@ def preconfigure_relationship(context, graph, node_instance):
@workflow(suffix_template='{node_instance.id}')
def postconfigure_relationship(context, graph, node_instance):
+ """
+
+ :param context:
+ :param graph:
+ :param node_instance:
+ :return:
+ """
graph.chain(tasks=relationships_tasks(
operation_name='aria.interfaces.relationship_lifecycle.postconfigure',
context=context,
@@ -81,6 +99,13 @@ def postconfigure_relationship(context, graph, node_instance):
@workflow(suffix_template='{node_instance.id}')
def establish_relationship(context, graph, node_instance):
+ """
+
+ :param context:
+ :param graph:
+ :param node_instance:
+ :return:
+ """
graph.chain(tasks=relationships_tasks(
operation_name='aria.interfaces.relationship_lifecycle.establish',
context=context,
@@ -91,6 +116,13 @@ def establish_relationship(context, graph, node_instance):
@workflow(suffix_template='{node_instance.id}')
def uninstall_node_instance(graph, context, node_instance):
+ """
+ A workflow which uninstalls a node instance.
+ :param WorkflowContext context: the workflow context
+ :param TaskGraph graph: the tasks graph of which to edit
+ :param node_instance: the node instance to uninstall
+ :return:
+ """
stop_node_instance = context.operation(
name='aria.interfaces.lifecycle.stop.{0}'.format(node_instance.id),
operation_details=node_instance.node.operations[
@@ -113,7 +145,14 @@ def uninstall_node_instance(graph, context, node_instance):
@workflow(suffix_template='{node_instance.id}')
def unlink_relationship(context, graph, node_instance):
- tasks=relationships_tasks(
+ """
+
+ :param context:
+ :param graph:
+ :param node_instance:
+ :return:
+ """
+ tasks = relationships_tasks(
operation_name='aria.interfaces.relationship_lifecycle.unlink',
context=context,
node_instance=node_instance
@@ -130,6 +169,16 @@ def execute_operation_on_instance(
operation,
operation_kwargs,
allow_kwargs_override):
+ """
+ A workflow which executes a single operation
+ :param WorkflowContext context: the workflow to execute.
+ :param TaskGraph graph: the tasks graph of which to edit
+ :param node_instance: the node instance to install
+ :param basestring operation: the operation name
+ :param dict operation_kwargs:
+ :param bool allow_kwargs_override:
+ :return:
+ """
if allow_kwargs_override is not None:
operation_kwargs['allow_kwargs_override'] = allow_kwargs_override
@@ -138,15 +187,23 @@ 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)
+ graph.add_task(
+ context.operation(
+ name=task_name,
+ operation_details=node_instance.node.operations[operation],
+ node_instance=node_instance,
+ parameters=operation_kwargs)
)
def relationships_tasks(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.
+ :param WorkflowContext context:
+ :param NodeInstance node_instance:
+ :return:
+ """
relationships_groups = groupby(
node_instance.relationship_instances,
key=lambda relationship_instance: relationship_instance.relationship.target_id)
@@ -165,6 +222,15 @@ def relationships_tasks(operation_name, context, node_instance):
def relationship_tasks(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
+ :param RelationshipInstance relationship_instance: the relationship instance itself
+ :param WorkflowContext context:
+ :param operation_name:
+ :param index: the relationship index - enables pretty print
+ :return:
+ """
index = index or node_instance.relationship_instances.index(relationship_instance)
sub_workflow_name = '{name}.{index}.{node_instance.id}'.format(
name=operation_name,