Convergence: Refactor worker

Refactor the worker service; move the check resource code to its own
class in another file and keep the convergence worker RPC API clean.

This refactor will help us contain the convergence logic in a separate
class file instead of in RCP API. The RPC service class should only have
the APIs it implements.

Change-Id: Ie9cf4daba7e6bf61f4cac3388494e8c9efefa4d7
changes/59/295159/16
Anant Patil 7 years ago committed by Rakesh H S
parent 5672c4649e
commit 829e80d06e
  1. 339
      heat/engine/check_resource.py
  2. 309
      heat/engine/worker.py
  3. 649
      heat/tests/engine/test_check_resource.py
  4. 621
      heat/tests/engine/test_engine_worker.py

@ -0,0 +1,339 @@
# Copyright (c) 2016 Hewlett-Packard Development Company, L.P.
#
# Licensed 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 six
from oslo_log import log as logging
from heat.common import exception
from heat.common.i18n import _LI
from heat.engine import resource
from heat.engine import scheduler
from heat.engine import stack as parser
from heat.engine import sync_point
from heat.objects import resource as resource_objects
from heat.rpc import listener_client
LOG = logging.getLogger(__name__)
class CheckResource(object):
def __init__(self,
engine_id,
rpc_client,
thread_group_mgr):
self.engine_id = engine_id
self._rpc_client = rpc_client
self.thread_group_mgr = thread_group_mgr
def _try_steal_engine_lock(self, cnxt, resource_id):
rs_obj = resource_objects.Resource.get_obj(cnxt,
resource_id)
if rs_obj.engine_id not in (None, self.engine_id):
if not listener_client.EngineListenerClient(
rs_obj.engine_id).is_alive(cnxt):
# steal the lock.
rs_obj.update_and_save({'engine_id': None})
return True
return False
def _trigger_rollback(self, stack):
LOG.info(_LI("Triggering rollback of %(stack_name)s %(action)s "),
{'action': stack.action, 'stack_name': stack.name})
stack.rollback()
def _handle_failure(self, cnxt, stack, failure_reason):
updated = stack.state_set(stack.action, stack.FAILED, failure_reason)
if not updated:
return False
if (not stack.disable_rollback and
stack.action in (stack.CREATE, stack.ADOPT, stack.UPDATE)):
self._trigger_rollback(stack)
else:
stack.purge_db()
return True
def _handle_resource_failure(self, cnxt, is_update, rsrc_id,
stack, failure_reason):
failure_handled = self._handle_failure(cnxt, stack, failure_reason)
if not failure_handled:
# Another concurrent update has taken over. But there is a
# possibility for that update to be waiting for this rsrc to
# complete, hence retrigger current rsrc for latest traversal.
traversal = stack.current_traversal
latest_stack = parser.Stack.load(cnxt, stack_id=stack.id)
if traversal != latest_stack.current_traversal:
self._retrigger_check_resource(cnxt, is_update, rsrc_id,
latest_stack)
def _handle_stack_timeout(self, cnxt, stack):
failure_reason = u'Timed out'
self._handle_failure(cnxt, stack, failure_reason)
def _do_check_resource(self, cnxt, current_traversal, tmpl, resource_data,
is_update, rsrc, stack, adopt_stack_data):
try:
if is_update:
try:
check_resource_update(rsrc, tmpl.id, resource_data,
self.engine_id,
stack)
except exception.UpdateReplace:
new_res_id = rsrc.make_replacement(tmpl.id)
LOG.info("Replacing resource with new id %s", new_res_id)
rpc_data = sync_point.serialize_input_data(resource_data)
self._rpc_client.check_resource(cnxt,
new_res_id,
current_traversal,
rpc_data, is_update,
adopt_stack_data)
return False
else:
check_resource_cleanup(rsrc, tmpl.id, resource_data,
self.engine_id, stack.time_remaining())
return True
except exception.UpdateInProgress:
if self._try_steal_engine_lock(cnxt, rsrc.id):
rpc_data = sync_point.serialize_input_data(resource_data)
# set the resource state as failed
status_reason = ('Worker went down '
'during resource %s' % rsrc.action)
rsrc.state_set(rsrc.action,
rsrc.FAILED,
six.text_type(status_reason))
self._rpc_client.check_resource(cnxt,
rsrc.id,
current_traversal,
rpc_data, is_update,
adopt_stack_data)
except exception.ResourceFailure as ex:
reason = 'Resource %s failed: %s' % (rsrc.action,
six.text_type(ex))
self._handle_resource_failure(cnxt, is_update, rsrc.id,
stack, reason)
except scheduler.Timeout:
# reload the stack to verify current traversal
stack = parser.Stack.load(cnxt, stack_id=stack.id)
if stack.current_traversal != current_traversal:
return
self._handle_stack_timeout(cnxt, stack)
return False
def _retrigger_check_resource(self, cnxt, is_update, resource_id, stack):
current_traversal = stack.current_traversal
graph = stack.convergence_dependencies.graph()
key = (resource_id, is_update)
if is_update:
# When re-triggering for a rsrc, we need to first check if update
# traversal is present for the rsrc in latest stack traversal,
# if No, then latest traversal is waiting for delete.
if (resource_id, is_update) not in graph:
key = (resource_id, not is_update)
LOG.info('Re-trigger resource: (%s, %s)' % (key[0], key[1]))
predecessors = set(graph[key])
try:
propagate_check_resource(cnxt, self._rpc_client, resource_id,
current_traversal, predecessors, key,
None, key[1], None)
except exception.EntityNotFound as e:
if e.entity != "Sync Point":
raise
def _initiate_propagate_resource(self, cnxt, resource_id,
current_traversal, is_update, rsrc,
stack):
deps = stack.convergence_dependencies
graph = deps.graph()
graph_key = (resource_id, is_update)
if graph_key not in graph and rsrc.replaces is not None:
# If we are a replacement, impersonate the replaced resource for
# the purposes of calculating whether subsequent resources are
# ready, since everybody has to work from the same version of the
# graph. Our real resource ID is sent in the input_data, so the
# dependencies will get updated to point to this resource in time
# for the next traversal.
graph_key = (rsrc.replaces, is_update)
def _get_input_data(req, fwd):
if fwd:
return construct_input_data(rsrc, stack)
else:
# Don't send data if initiating clean-up for self i.e.
# initiating delete of a replaced resource
if req not in graph_key:
# send replaced resource as needed_by if it exists
return (rsrc.replaced_by
if rsrc.replaced_by is not None
else resource_id)
return None
try:
for req, fwd in deps.required_by(graph_key):
input_data = _get_input_data(req, fwd)
propagate_check_resource(
cnxt, self._rpc_client, req, current_traversal,
set(graph[(req, fwd)]), graph_key, input_data, fwd,
stack.adopt_stack_data)
check_stack_complete(cnxt, stack, current_traversal,
resource_id, deps, is_update)
except exception.EntityNotFound as e:
if e.entity == "Sync Point":
# Reload the stack to determine the current traversal, and
# check the SyncPoint for the current node to determine if
# it is ready. If it is, then retrigger the current node
# with the appropriate data for the latest traversal.
stack = parser.Stack.load(cnxt, stack_id=rsrc.stack.id)
if current_traversal == stack.current_traversal:
LOG.debug('[%s] Traversal sync point missing.',
current_traversal)
return
self._retrigger_check_resource(cnxt, is_update,
resource_id, stack)
else:
raise
def check(self, cnxt, resource_id, current_traversal,
resource_data, is_update, adopt_stack_data,
rsrc, stack):
"""Process a node in the dependency graph.
The node may be associated with either an update or a cleanup of its
associated resource.
"""
if stack.has_timed_out():
self._handle_stack_timeout(cnxt, stack)
return
tmpl = stack.t
stack.adopt_stack_data = adopt_stack_data
stack.thread_group_mgr = self.thread_group_mgr
if is_update:
if (rsrc.replaced_by is not None and
rsrc.current_template_id != tmpl.id):
return
check_resource_done = self._do_check_resource(cnxt, current_traversal,
tmpl, resource_data,
is_update,
rsrc, stack,
adopt_stack_data)
if check_resource_done:
# initiate check on next set of resources from graph
self._initiate_propagate_resource(cnxt, resource_id,
current_traversal, is_update,
rsrc, stack)
def load_resource(cnxt, resource_id, resource_data, is_update):
if is_update:
cache_data = {in_data.get(
'name'): in_data for in_data in resource_data.values()
if in_data is not None}
else:
# no data to resolve in cleanup phase
cache_data = {}
try:
return resource.Resource.load(cnxt, resource_id,
is_update, cache_data)
except (exception.ResourceNotFound, exception.NotFound):
# can be ignored
return None, None, None
def construct_input_data(rsrc, curr_stack):
attributes = curr_stack.get_dep_attrs(
six.itervalues(curr_stack.resources),
curr_stack.outputs,
rsrc.name)
resolved_attributes = {}
for attr in attributes:
try:
if isinstance(attr, six.string_types):
resolved_attributes[attr] = rsrc.get_attribute(attr)
else:
resolved_attributes[attr] = rsrc.get_attribute(*attr)
except exception.InvalidTemplateAttribute as ita:
LOG.info(six.text_type(ita))
input_data = {'id': rsrc.id,
'name': rsrc.name,
'reference_id': rsrc.get_reference_id(),
'attrs': resolved_attributes,
'status': rsrc.status,
'action': rsrc.action,
'uuid': rsrc.uuid}
return input_data
def check_stack_complete(cnxt, stack, current_traversal, sender_id, deps,
is_update):
"""Mark the stack complete if the update is complete.
Complete is currently in the sense that all desired resources are in
service, not that superfluous ones have been cleaned up.
"""
roots = set(deps.roots())
if (sender_id, is_update) not in roots:
return
def mark_complete(stack_id, data):
stack.mark_complete()
sender_key = (sender_id, is_update)
sync_point.sync(cnxt, stack.id, current_traversal, True,
mark_complete, roots, {sender_key: None})
def propagate_check_resource(cnxt, rpc_client, next_res_id,
current_traversal, predecessors, sender_key,
sender_data, is_update, adopt_stack_data):
"""Trigger processing of node if all of its dependencies are satisfied."""
def do_check(entity_id, data):
rpc_client.check_resource(cnxt, entity_id, current_traversal,
data, is_update, adopt_stack_data)
sync_point.sync(cnxt, next_res_id, current_traversal,
is_update, do_check, predecessors,
{sender_key: sender_data})
def check_resource_update(rsrc, template_id, resource_data, engine_id,
stack):
"""Create or update the Resource if appropriate."""
if rsrc.action == resource.Resource.INIT:
rsrc.create_convergence(template_id, resource_data, engine_id,
stack.time_remaining())
else:
rsrc.update_convergence(template_id, resource_data, engine_id,
stack.time_remaining(), stack)
def check_resource_cleanup(rsrc, template_id, resource_data, engine_id,
timeout):
"""Delete the Resource if appropriate."""
rsrc.delete_convergence(template_id, resource_data, engine_id, timeout)

@ -17,19 +17,13 @@ from oslo_log import log as logging
import oslo_messaging
from oslo_service import service
from osprofiler import profiler
import six
from heat.common import context
from heat.common import exception
from heat.common.i18n import _LE
from heat.common.i18n import _LI
from heat.common import messaging as rpc_messaging
from heat.engine import resource
from heat.engine import scheduler
from heat.engine import stack as parser
from heat.engine import check_resource
from heat.engine import sync_point
from heat.objects import resource as resource_objects
from heat.rpc import listener_client
from heat.rpc import worker_client as rpc_client
LOG = logging.getLogger(__name__)
@ -92,200 +86,6 @@ class WorkerService(service.Service):
super(WorkerService, self).stop()
def _try_steal_engine_lock(self, cnxt, resource_id):
rs_obj = resource_objects.Resource.get_obj(cnxt,
resource_id)
if (rs_obj.engine_id != self.engine_id and
rs_obj.engine_id is not None):
if not listener_client.EngineListenerClient(
rs_obj.engine_id).is_alive(cnxt):
# steal the lock.
rs_obj.update_and_save({'engine_id': None})
return True
return False
def _trigger_rollback(self, stack):
LOG.info(_LI("Triggering rollback of %(stack_name)s %(action)s "),
{'action': stack.action, 'stack_name': stack.name})
stack.rollback()
def _handle_failure(self, cnxt, stack, failure_reason):
updated = stack.state_set(stack.action, stack.FAILED, failure_reason)
if not updated:
return False
if (not stack.disable_rollback and
stack.action in (stack.CREATE, stack.ADOPT, stack.UPDATE)):
self._trigger_rollback(stack)
else:
stack.purge_db()
return True
def _handle_resource_failure(self, cnxt, is_update, rsrc_id,
stack, failure_reason):
failure_handled = self._handle_failure(cnxt, stack, failure_reason)
if not failure_handled:
# Another concurrent update has taken over. But there is a
# possibility for that update to be waiting for this rsrc to
# complete, hence retrigger current rsrc for latest traversal.
traversal = stack.current_traversal
latest_stack = parser.Stack.load(cnxt, stack_id=stack.id)
if traversal != latest_stack.current_traversal:
self._retrigger_check_resource(cnxt, is_update, rsrc_id,
latest_stack)
def _handle_stack_timeout(self, cnxt, stack):
failure_reason = u'Timed out'
self._handle_failure(cnxt, stack, failure_reason)
def _load_resource(self, cnxt, resource_id, resource_data, is_update):
if is_update:
cache_data = {in_data.get(
'name'): in_data for in_data in resource_data.values()
if in_data is not None}
else:
# no data to resolve in cleanup phase
cache_data = {}
try:
return resource.Resource.load(cnxt, resource_id,
is_update, cache_data)
except (exception.ResourceNotFound, exception.NotFound):
pass # can be ignored
return None, None, None
def _do_check_resource(self, cnxt, current_traversal, tmpl, resource_data,
is_update, rsrc, stack, adopt_stack_data):
try:
if is_update:
try:
check_resource_update(rsrc, tmpl.id, resource_data,
self.engine_id,
stack)
except exception.UpdateReplace:
new_res_id = rsrc.make_replacement(tmpl.id)
LOG.info("Replacing resource with new id %s", new_res_id)
rpc_data = sync_point.serialize_input_data(resource_data)
self._rpc_client.check_resource(cnxt,
new_res_id,
current_traversal,
rpc_data, is_update,
adopt_stack_data)
return False
else:
check_resource_cleanup(rsrc, tmpl.id, resource_data,
self.engine_id, stack.time_remaining())
return True
except exception.UpdateInProgress:
if self._try_steal_engine_lock(cnxt, rsrc.id):
rpc_data = sync_point.serialize_input_data(resource_data)
# set the resource state as failed
status_reason = ('Worker went down '
'during resource %s' % rsrc.action)
rsrc.state_set(rsrc.action,
rsrc.FAILED,
six.text_type(status_reason))
self._rpc_client.check_resource(cnxt,
rsrc.id,
current_traversal,
rpc_data, is_update,
adopt_stack_data)
except exception.ResourceFailure as ex:
reason = 'Resource %s failed: %s' % (rsrc.action,
six.text_type(ex))
self._handle_resource_failure(cnxt, is_update, rsrc.id,
stack, reason)
except scheduler.Timeout:
# reload the stack to verify current traversal
stack = parser.Stack.load(cnxt, stack_id=stack.id)
if stack.current_traversal != current_traversal:
return
self._handle_stack_timeout(cnxt, stack)
return False
def _retrigger_check_resource(self, cnxt, is_update, resource_id, stack):
current_traversal = stack.current_traversal
graph = stack.convergence_dependencies.graph()
key = (resource_id, is_update)
if is_update:
# When re-triggering for a rsrc, we need to first check if update
# traversal is present for the rsrc in latest stack traversal,
# if No, then latest traversal is waiting for delete.
if (resource_id, is_update) not in graph:
key = (resource_id, not is_update)
LOG.info('Re-trigger resource: (%s, %s)' % (key[0], key[1]))
predecessors = set(graph[key])
try:
propagate_check_resource(cnxt, self._rpc_client, resource_id,
current_traversal, predecessors, key,
None, key[1], None)
except exception.EntityNotFound as e:
if e.entity == "Sync Point":
pass
else:
raise
def _initiate_propagate_resource(self, cnxt, resource_id,
current_traversal, is_update, rsrc,
stack):
deps = stack.convergence_dependencies
graph = deps.graph()
graph_key = (resource_id, is_update)
if graph_key not in graph and rsrc.replaces is not None:
# If we are a replacement, impersonate the replaced resource for
# the purposes of calculating whether subsequent resources are
# ready, since everybody has to work from the same version of the
# graph. Our real resource ID is sent in the input_data, so the
# dependencies will get updated to point to this resource in time
# for the next traversal.
graph_key = (rsrc.replaces, is_update)
def _get_input_data(req, fwd):
if fwd:
return construct_input_data(rsrc, stack)
else:
# Don't send data if initiating clean-up for self i.e.
# initiating delete of a replaced resource
if req not in graph_key:
# send replaced resource as needed_by if it exists
return (rsrc.replaced_by
if rsrc.replaced_by is not None
else resource_id)
return None
try:
for req, fwd in deps.required_by(graph_key):
input_data = _get_input_data(req, fwd)
propagate_check_resource(
cnxt, self._rpc_client, req, current_traversal,
set(graph[(req, fwd)]), graph_key, input_data, fwd,
stack.adopt_stack_data)
check_stack_complete(cnxt, stack, current_traversal,
resource_id, deps, is_update)
except exception.EntityNotFound as e:
if e.entity == "Sync Point":
# Reload the stack to determine the current traversal, and
# check the SyncPoint for the current node to determine if
# it is ready. If it is, then retrigger the current node
# with the appropriate data for the latest traversal.
stack = parser.Stack.load(cnxt, stack_id=rsrc.stack.id)
if current_traversal == stack.current_traversal:
LOG.debug('[%s] Traversal sync point missing.',
current_traversal)
return
self._retrigger_check_resource(cnxt, is_update,
resource_id, stack)
else:
raise
@context.request_context
def check_resource(self, cnxt, resource_id, current_traversal, data,
is_update, adopt_stack_data):
@ -295,9 +95,8 @@ class WorkerService(service.Service):
associated resource.
"""
resource_data = dict(sync_point.deserialize_input_data(data))
rsrc, rsrc_owning_stack, stack = self._load_resource(cnxt, resource_id,
resource_data,
is_update)
rsrc, rsrc_owning_stack, stack = check_resource.load_resource(
cnxt, resource_id, resource_data, is_update)
if rsrc is None:
return
@ -306,102 +105,8 @@ class WorkerService(service.Service):
LOG.debug('[%s] Traversal cancelled; stopping.', current_traversal)
return
if stack.has_timed_out():
self._handle_stack_timeout(cnxt, stack)
return
tmpl = stack.t
stack.adopt_stack_data = adopt_stack_data
stack.thread_group_mgr = self.thread_group_mgr
if is_update:
if (rsrc.replaced_by is not None and
rsrc.current_template_id != tmpl.id):
return
check_resource_done = self._do_check_resource(cnxt, current_traversal,
tmpl, resource_data,
is_update,
rsrc, stack,
adopt_stack_data)
if check_resource_done:
# initiate check on next set of resources from graph
self._initiate_propagate_resource(cnxt, resource_id,
current_traversal, is_update,
rsrc, stack)
def construct_input_data(rsrc, curr_stack):
attributes = curr_stack.get_dep_attrs(
six.itervalues(curr_stack.resources),
curr_stack.outputs,
rsrc.name)
resolved_attributes = {}
for attr in attributes:
try:
if isinstance(attr, six.string_types):
resolved_attributes[attr] = rsrc.get_attribute(attr)
else:
resolved_attributes[attr] = rsrc.get_attribute(*attr)
except exception.InvalidTemplateAttribute as ita:
LOG.info(six.text_type(ita))
input_data = {'id': rsrc.id,
'name': rsrc.name,
'reference_id': rsrc.get_reference_id(),
'attrs': resolved_attributes,
'status': rsrc.status,
'action': rsrc.action,
'uuid': rsrc.uuid}
return input_data
def check_stack_complete(cnxt, stack, current_traversal, sender_id, deps,
is_update):
"""Mark the stack complete if the update is complete.
Complete is currently in the sense that all desired resources are in
service, not that superfluous ones have been cleaned up.
"""
roots = set(deps.roots())
if (sender_id, is_update) not in roots:
return
def mark_complete(stack_id, data):
stack.mark_complete()
sender_key = (sender_id, is_update)
sync_point.sync(cnxt, stack.id, current_traversal, True,
mark_complete, roots, {sender_key: None})
def propagate_check_resource(cnxt, rpc_client, next_res_id,
current_traversal, predecessors, sender_key,
sender_data, is_update, adopt_stack_data):
"""Trigger processing of node if all of its dependencies are satisfied."""
def do_check(entity_id, data):
rpc_client.check_resource(cnxt, entity_id, current_traversal,
data, is_update, adopt_stack_data)
sync_point.sync(cnxt, next_res_id, current_traversal,
is_update, do_check, predecessors,
{sender_key: sender_data})
def check_resource_update(rsrc, template_id, resource_data, engine_id,
stack):
"""Create or update the Resource if appropriate."""
if rsrc.action == resource.Resource.INIT:
rsrc.create_convergence(template_id, resource_data, engine_id,
stack.time_remaining())
else:
rsrc.update_convergence(template_id, resource_data, engine_id,
stack.time_remaining(), stack)
cr = check_resource.CheckResource(self.engine_id, self._rpc_client,
self.thread_group_mgr)
def check_resource_cleanup(rsrc, template_id, resource_data, engine_id,
timeout):
"""Delete the Resource if appropriate."""
rsrc.delete_convergence(template_id, resource_data, engine_id, timeout)
cr.check(cnxt, resource_id, current_traversal, resource_data,
is_update, adopt_stack_data, rsrc, stack)

@ -0,0 +1,649 @@
# Copyright (c) 2016 Hewlett-Packard Development Company, L.P.
#
# Licensed 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 mock
from oslo_config import cfg
from heat.common import exception
from heat.engine import check_resource
from heat.engine import dependencies
from heat.engine import resource
from heat.engine import scheduler
from heat.engine import stack
from heat.engine import sync_point
from heat.engine import worker
from heat.rpc import worker_client
from heat.tests import common
from heat.tests.engine import tools
from heat.tests import utils
@mock.patch.object(check_resource, 'construct_input_data')
@mock.patch.object(check_resource, 'check_stack_complete')
@mock.patch.object(check_resource, 'propagate_check_resource')
@mock.patch.object(check_resource, 'check_resource_cleanup')
@mock.patch.object(check_resource, 'check_resource_update')
class CheckWorkflowUpdateTest(common.HeatTestCase):
@mock.patch.object(worker_client.WorkerClient, 'check_resource',
lambda *_: None)
def setUp(self):
super(CheckWorkflowUpdateTest, self).setUp()
thread_group_mgr = mock.Mock()
cfg.CONF.set_default('convergence_engine', True)
self.worker = worker.WorkerService('host-1',
'topic-1',
'engine_id',
thread_group_mgr)
self.cr = check_resource.CheckResource(self.worker.engine_id,
self.worker._rpc_client,
self.worker.thread_group_mgr)
self.worker._rpc_client = worker_client.WorkerClient()
self.ctx = utils.dummy_context()
self.stack = tools.get_stack(
'check_workflow_create_stack', self.ctx,
template=tools.string_template_five, convergence=True)
self.stack.converge_stack(self.stack.t)
self.resource = self.stack['A']
self.is_update = True
self.graph_key = (self.resource.id, self.is_update)
self.orig_load_method = stack.Stack.load
stack.Stack.load = mock.Mock(return_value=self.stack)
def tearDown(self):
super(CheckWorkflowUpdateTest, self).tearDown()
stack.Stack.load = self.orig_load_method
def test_resource_not_available(
self, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.worker.check_resource(
self.ctx, 'non-existant-id', self.stack.current_traversal, {},
True, None)
for mocked in [mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid]:
self.assertFalse(mocked.called)
def test_stale_traversal(
self, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.worker.check_resource(self.ctx, self.resource.id,
'stale-traversal', {}, True, None)
for mocked in [mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid]:
self.assertFalse(mocked.called)
def test_is_update_traversal(
self, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.worker.check_resource(
self.ctx, self.resource.id, self.stack.current_traversal, {},
self.is_update, None)
mock_cru.assert_called_once_with(self.resource,
self.resource.stack.t.id,
{}, self.worker.engine_id,
mock.ANY)
self.assertFalse(mock_crc.called)
expected_calls = []
for req, fwd in self.stack.convergence_dependencies.leaves():
expected_calls.append(
(mock.call.worker.propagate_check_resource.
assert_called_once_with(
self.ctx, mock.ANY, mock.ANY,
self.stack.current_traversal, mock.ANY,
self.graph_key, {}, self.is_update)))
mock_csc.assert_called_once_with(
self.ctx, mock.ANY, self.stack.current_traversal,
self.resource.id,
mock.ANY, True)
@mock.patch.object(resource.Resource, 'make_replacement')
@mock.patch.object(stack.Stack, 'time_remaining')
def test_is_update_traversal_raise_update_replace(
self, tr, mock_mr, mock_cru, mock_crc, mock_pcr, mock_csc,
mock_cid):
mock_cru.side_effect = exception.UpdateReplace
tr.return_value = 317
self.worker.check_resource(
self.ctx, self.resource.id, self.stack.current_traversal, {},
self.is_update, None)
mock_cru.assert_called_once_with(self.resource,
self.resource.stack.t.id,
{}, self.worker.engine_id,
mock.ANY)
self.assertTrue(mock_mr.called)
self.assertFalse(mock_crc.called)
self.assertFalse(mock_pcr.called)
self.assertFalse(mock_csc.called)
@mock.patch.object(check_resource.CheckResource, '_try_steal_engine_lock')
@mock.patch.object(stack.Stack, 'time_remaining')
@mock.patch.object(resource.Resource, 'state_set')
def test_is_update_traversal_raise_update_inprogress(
self, mock_ss, tr, mock_tsl, mock_cru, mock_crc, mock_pcr,
mock_csc, mock_cid):
mock_cru.side_effect = exception.UpdateInProgress
self.worker.engine_id = 'some-thing-else'
mock_tsl.return_value = True
tr.return_value = 317
self.worker.check_resource(
self.ctx, self.resource.id, self.stack.current_traversal, {},
self.is_update, None)
mock_cru.assert_called_once_with(self.resource,
self.resource.stack.t.id,
{}, self.worker.engine_id,
mock.ANY)
mock_ss.assert_called_once_with(self.resource.action,
resource.Resource.FAILED,
mock.ANY)
self.assertFalse(mock_crc.called)
self.assertFalse(mock_pcr.called)
self.assertFalse(mock_csc.called)
def test_try_steal_lock_alive(
self, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
res = self.cr._try_steal_engine_lock(self.ctx,
self.resource.id)
self.assertFalse(res)
@mock.patch.object(check_resource.listener_client, 'EngineListenerClient')
@mock.patch.object(check_resource.resource_objects.Resource, 'get_obj')
def test_try_steal_lock_dead(
self, mock_get, mock_elc, mock_cru, mock_crc, mock_pcr,
mock_csc, mock_cid):
fake_res = mock.Mock()
fake_res.engine_id = 'some-thing-else'
mock_get.return_value = fake_res
mock_elc.return_value.is_alive.return_value = False
res = self.cr._try_steal_engine_lock(self.ctx,
self.resource.id)
self.assertTrue(res)
@mock.patch.object(check_resource.listener_client, 'EngineListenerClient')
@mock.patch.object(check_resource.resource_objects.Resource, 'get_obj')
def test_try_steal_lock_not_dead(
self, mock_get, mock_elc, mock_cru, mock_crc, mock_pcr,
mock_csc, mock_cid):
fake_res = mock.Mock()
fake_res.engine_id = self.worker.engine_id
mock_get.return_value = fake_res
mock_elc.return_value.is_alive.return_value = True
res = self.cr._try_steal_engine_lock(self.ctx, self.resource.id)
self.assertFalse(res)
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_resource_update_failure_sets_stack_state_as_failed(
self, mock_tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.stack.state_set(self.stack.UPDATE, self.stack.IN_PROGRESS, '')
self.resource.state_set(self.resource.UPDATE,
self.resource.IN_PROGRESS)
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_cru.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.resource.UPDATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
s = self.stack.load(self.ctx, stack_id=self.stack.id)
self.assertEqual((s.UPDATE, s.FAILED), (s.action, s.status))
self.assertEqual('Resource UPDATE failed: '
'ResourceNotAvailable: resources.A: The Resource (A)'
' is not available.', s.status_reason)
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_resource_cleanup_failure_sets_stack_state_as_failed(
self, mock_tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.is_update = False # invokes check_resource_cleanup
self.stack.state_set(self.stack.UPDATE, self.stack.IN_PROGRESS, '')
self.resource.state_set(self.resource.UPDATE,
self.resource.IN_PROGRESS)
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_crc.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.resource.UPDATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
s = self.stack.load(self.ctx, stack_id=self.stack.id)
self.assertEqual((s.UPDATE, s.FAILED), (s.action, s.status))
self.assertEqual('Resource UPDATE failed: '
'ResourceNotAvailable: resources.A: The Resource (A)'
' is not available.', s.status_reason)
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_resource_update_failure_triggers_rollback_if_enabled(
self, mock_tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.stack.disable_rollback = False
self.stack.store()
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_cru.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.resource.UPDATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
self.assertTrue(mock_tr.called)
# make sure the rollback is called on given stack
call_args, call_kwargs = mock_tr.call_args
called_stack = call_args[0]
self.assertEqual(self.stack.id, called_stack.id)
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_resource_cleanup_failure_triggers_rollback_if_enabled(
self, mock_tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.is_update = False # invokes check_resource_cleanup
self.stack.disable_rollback = False
self.stack.store()
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_crc.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.resource.UPDATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
self.assertTrue(mock_tr.called)
# make sure the rollback is called on given stack
call_args, call_kwargs = mock_tr.call_args
called_stack = call_args[0]
self.assertEqual(self.stack.id, called_stack.id)
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_rollback_is_not_triggered_on_rollback_disabled_stack(
self, mock_tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.stack.disable_rollback = True
self.stack.store()
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_cru.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.stack.CREATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
self.assertFalse(mock_tr.called)
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_rollback_not_re_triggered_for_a_rolling_back_stack(
self, mock_tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.stack.disable_rollback = False
self.stack.action = self.stack.ROLLBACK
self.stack.status = self.stack.IN_PROGRESS
self.stack.store()
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_cru.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.stack.CREATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
self.assertFalse(mock_tr.called)
def test_resource_update_failure_purges_db_for_stack_failure(
self, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.stack.disable_rollback = True
self.stack.store()
self.stack.purge_db = mock.Mock()
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_cru.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.resource.UPDATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
self.assertTrue(self.stack.purge_db.called)
def test_resource_cleanup_failure_purges_db_for_stack_failure(
self, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
self.is_update = False
self.stack.disable_rollback = True
self.stack.store()
self.stack.purge_db = mock.Mock()
dummy_ex = exception.ResourceNotAvailable(
resource_name=self.resource.name)
mock_crc.side_effect = exception.ResourceFailure(
dummy_ex, self.resource, action=self.resource.UPDATE)
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal, {},
self.is_update, None)
self.assertTrue(self.stack.purge_db.called)
@mock.patch.object(check_resource.CheckResource,
'_retrigger_check_resource')
@mock.patch.object(stack.Stack, 'load')
def test_initiate_propagate_rsrc_retriggers_check_rsrc_on_new_stack_update(
self, mock_stack_load, mock_rcr, mock_cru, mock_crc, mock_pcr,
mock_csc, mock_cid):
key = sync_point.make_key(self.resource.id,
self.stack.current_traversal,
self.is_update)
mock_pcr.side_effect = exception.EntityNotFound(entity='Sync Point',
name=key)
updated_stack = stack.Stack(self.ctx, self.stack.name, self.stack.t,
self.stack.id,
current_traversal='some_newy_trvl_uuid')
mock_stack_load.return_value = updated_stack
self.cr._initiate_propagate_resource(self.ctx, self.resource.id,
self.stack.current_traversal,
self.is_update, self.resource,
self.stack)
mock_rcr.assert_called_once_with(self.ctx, self.is_update,
self.resource.id, updated_stack)
@mock.patch.object(sync_point, 'sync')
def test_retrigger_check_resource(self, mock_sync, mock_cru, mock_crc,
mock_pcr, mock_csc, mock_cid):
resC = self.stack['C']
# A, B are predecessors to C when is_update is True
expected_predecessors = {(self.stack['A'].id, True),
(self.stack['B'].id, True)}
self.cr._retrigger_check_resource(self.ctx, self.is_update,
resC.id, self.stack)
mock_pcr.assert_called_once_with(self.ctx, mock.ANY, resC.id,
self.stack.current_traversal,
mock.ANY, (resC.id, True), None,
True, None)
call_args, call_kwargs = mock_pcr.call_args
actual_predecessors = call_args[4]
self.assertItemsEqual(expected_predecessors, actual_predecessors)
def test_retrigger_check_resource_new_traversal_deletes_rsrc(
self, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
# mock dependencies to indicate a rsrc with id 2 is not present
# in latest traversal
self.stack._convg_deps = dependencies.Dependencies([
[(1, False), (1, True)], [(2, False), None]])
# simulate rsrc 2 completing its update for old traversal
# and calling rcr
self.cr._retrigger_check_resource(self.ctx, True, 2, self.stack)
# Ensure that pcr was called with proper delete traversal
mock_pcr.assert_called_once_with(self.ctx, mock.ANY, 2,
self.stack.current_traversal,
mock.ANY, (2, False), None,
False, None)
@mock.patch.object(stack.Stack, 'purge_db')
def test_handle_failure(self, mock_purgedb, mock_cru, mock_crc, mock_pcr,
mock_csc, mock_cid):
self.cr._handle_failure(self.ctx, self.stack, 'dummy-reason')
mock_purgedb.assert_called_once_with()
self.assertEqual('dummy-reason', self.stack.status_reason)
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_handle_failure_rollback(self, mock_tr, mock_cru, mock_crc,
mock_pcr, mock_csc, mock_cid):
self.stack.disable_rollback = False
self.stack.state_set(self.stack.UPDATE, self.stack.IN_PROGRESS, '')
self.cr._handle_failure(self.ctx, self.stack, 'dummy-reason')
mock_tr.assert_called_once_with(self.stack)
@mock.patch.object(stack.Stack, 'purge_db')
@mock.patch.object(stack.Stack, 'state_set')
@mock.patch.object(check_resource.CheckResource,
'_retrigger_check_resource')
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_handle_rsrc_failure_when_update_fails(
self, mock_tr, mock_rcr, mock_ss, mock_pdb, mock_cru, mock_crc,
mock_pcr, mock_csc, mock_cid):
# Emulate failure
mock_ss.return_value = False
self.cr._handle_resource_failure(self.ctx, self.is_update,
self.resource.id, self.stack,
'dummy-reason')
self.assertTrue(mock_ss.called)
self.assertFalse(mock_rcr.called)
self.assertFalse(mock_pdb.called)
self.assertFalse(mock_tr.called)
@mock.patch.object(stack.Stack, 'purge_db')
@mock.patch.object(stack.Stack, 'state_set')
@mock.patch.object(check_resource.CheckResource,
'_retrigger_check_resource')
@mock.patch.object(check_resource.CheckResource, '_trigger_rollback')
def test_handle_rsrc_failure_when_update_fails_different_traversal(
self, mock_tr, mock_rcr, mock_ss, mock_pdb, mock_cru, mock_crc,
mock_pcr, mock_csc, mock_cid):
# Emulate failure
mock_ss.return_value = False
# Emulate new traversal
new_stack = tools.get_stack('check_workflow_create_stack', self.ctx,
template=tools.string_template_five,
convergence=True)
new_stack.current_traversal = 'new_traversal'
stack.Stack.load = mock.Mock(return_value=new_stack)
self.cr._handle_resource_failure(self.ctx, self.is_update,
self.resource.id,
self.stack, 'dummy-reason')
# Ensure retrigger called
self.assertTrue(mock_rcr.called)
self.assertTrue(mock_ss.called)
self.assertFalse(mock_pdb.called)
self.assertFalse(mock_tr.called)
@mock.patch.object(check_resource.CheckResource, '_handle_failure')
def test_handle_stack_timeout(self, mock_hf, mock_cru, mock_crc, mock_pcr,
mock_csc, mock_cid):
self.cr._handle_stack_timeout(self.ctx, self.stack)
mock_hf.assert_called_once_with(self.ctx, self.stack, u'Timed out')
@mock.patch.object(check_resource.CheckResource,
'_handle_stack_timeout')
def test_do_check_resource_marks_stack_as_failed_if_stack_timesout(
self, mock_hst, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
mock_cru.side_effect = scheduler.Timeout(None, 60)
self.is_update = True
self.cr._do_check_resource(self.ctx, self.stack.current_traversal,
self.stack.t, {}, self.is_update,
self.resource, self.stack, {})
mock_hst.assert_called_once_with(self.ctx, self.stack)
@mock.patch.object(check_resource.CheckResource,
'_handle_stack_timeout')
def test_do_check_resource_ignores_timeout_for_new_update(
self, mock_hst, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
# Ensure current_traversal is check before marking the stack as
# failed due to time-out.
mock_cru.side_effect = scheduler.Timeout(None, 60)
self.is_update = True
old_traversal = self.stack.current_traversal
self.stack.current_traversal = 'new_traversal'
self.cr._do_check_resource(self.ctx, old_traversal,
self.stack.t, {}, self.is_update,
self.resource, self.stack, {})
self.assertFalse(mock_hst.called)
@mock.patch.object(stack.Stack, 'has_timed_out')
@mock.patch.object(check_resource.CheckResource,
'_handle_stack_timeout')
def test_check_resource_handles_timeout(self, mock_hst, mock_to, mock_cru,
mock_crc, mock_pcr, mock_csc,
mock_cid):
mock_to.return_value = True
self.worker.check_resource(self.ctx, self.resource.id,
self.stack.current_traversal,
{}, self.is_update, {})
self.assertTrue(mock_hst.called)
@mock.patch.object(check_resource, 'construct_input_data')
@mock.patch.object(check_resource, 'check_stack_complete')
@mock.patch.object(check_resource, 'propagate_check_resource')
@mock.patch.object(check_resource, 'check_resource_cleanup')
@mock.patch.object(check_resource, 'check_resource_update')
class CheckWorkflowCleanupTest(common.HeatTestCase):
@mock.patch.object(worker_client.WorkerClient, 'check_resource',
lambda *_: None)
def setUp(self):
super(CheckWorkflowCleanupTest, self).setUp()
thread_group_mgr = mock.Mock()
self.worker = worker.WorkerService('host-1',
'topic-1',
'engine_id',
thread_group_mgr)
self.worker._rpc_client = worker_client.WorkerClient()
self.ctx = utils.dummy_context()
tstack = tools.get_stack(
'check_workflow_create_stack', self.ctx,
template=tools.string_template_five, convergence=True)
tstack.converge_stack(tstack.t, action=tstack.CREATE)
self.stack = stack.Stack.load(self.ctx, stack_id=tstack.id)
self.stack.converge_stack(self.stack.t, action=self.stack.DELETE)
self.resource = self.stack['A']
self.is_update = False
self.graph_key = (self.resource.id, self.is_update)
@mock.patch.object(stack.Stack, 'time_remaining')
def test_is_cleanup_traversal(
self, tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
tr.return_value = 317
self.worker.check_resource(
self.ctx, self.resource.id, self.stack.current_traversal, {},
self.is_update, None)
self.assertFalse(mock_cru.called)
mock_crc.assert_called_once_with(
self.resource, self.resource.stack.t.id,
{}, self.worker.engine_id,
tr())
@mock.patch.object(stack.Stack, 'time_remaining')
def test_is_cleanup_traversal_raise_update_inprogress(
self, tr, mock_cru, mock_crc, mock_pcr, mock_csc, mock_cid):
mock_crc.side_effect = exception.UpdateInProgress
tr.return_value = 317
self.worker.check_resource(
self.ctx, self.resource.id, self.stack.current_traversal, {},
self.is_update, None)
mock_crc.assert_called_once_with(self.resource,
self.resource.stack.t.id,
{}, self.worker.engine_id,
tr())
self.assertFalse(mock_cru.called)
self.assertFalse(mock_pcr.called)
self.assertFalse(mock_csc.called)
class MiscMethodsTest(common.HeatTestCase):
def setUp(self):
super(MiscMethodsTest, self).setUp()
cfg.CONF.set_default('convergence_engine', True)
self.ctx = utils.dummy_context()
self.stack = tools.get_stack(
'check_workflow_create_stack', self.ctx,
template=tools.attr_cache_template, convergence=True)
self.stack.converge_stack(self.stack.t)
self.resource = self.stack['A']
def test_construct_input_data_ok(self):
expected_input_data = {'attrs': {(u'flat_dict', u'key2'): 'val2',
(u'flat_dict', u'key3'): 'val3',
(u'nested_dict', u'dict', u'a'): 1,
(u'nested_dict', u'dict', u'b'): 2},
'id': mock.ANY,
'reference_id': 'A',
'name': 'A',
'uuid': mock.ANY,
'action': mock.ANY,
'status': mock.ANY}
actual_input_data = check_resource.construct_input_data(self.resource,
self.stack)
self.assertEqual(expected_input_data, actual_input_data)
def test_construct_input_data_exception(self):
expected_input_data = {'attrs': {},
'id': mock.ANY,
'reference_id': 'A',
'name': 'A',
'uuid': mock.ANY,
'action': mock.ANY,
'status': mock.ANY}
self.resource.get_attribute = mock.Mock(
side_effect=exception.InvalidTemplateAttribute(resource='A',
key='value'))
actual_input_data = check_resource.construct_input_data(self.resource,
self.stack)
self.assertEqual(expected_input_data, actual_input_data)
@mock.patch.object(sync_point, 'sync')
def test_check_stack_complete_root(self, mock_sync):
check_resource.check_stack_complete(
self.ctx, self.stack, self.stack.current_traversal,
self.stack['E'].id, self.stack.convergence_dependencies,
True)
mock_sync.assert_called_once_with(
self.ctx, self.stack.id, self.stack.current_traversal, True,
mock.ANY, mock.ANY, {(self.stack['E'].id, True): None})
@mock.patch.object(sync_point, 'sync')
def test_check_stack_complete_child(self, mock_sync):
check_resource.check_stack_complete(
self.ctx, self.stack, self.stack.current_traversal,
self.resource.id, self.stack.convergence_dependencies,
True)
self.assertFalse(mock_sync.called)
@mock.patch.object(dependencies.Dependencies, 'roots')
@mock.patch.object(stack.Stack, '_persist_state')
def test_check_stack_complete_persist_called(self, mock_persist_state,
mock_dep_roots):
mock_dep_roots.return_value = [(1, True)]
check_resource.check_stack_complete(
self.ctx, self.stack, self.stack.current_traversal,
1, self.stack.convergence_dependencies,
True)
self.assertTrue(mock_persist_state.called)
@mock.patch.object(sync_point, 'sync')
def test_propagate_check_resource(self, mock_sync):
check_resource.propagate_check_resource(
self.ctx, mock.ANY, mock.ANY,
self.stack.current_traversal, mock.ANY,
('A', True), {}, True, None)
self.assertTrue(mock_sync.called)
@mock.patch.object(resource.Resource, 'create_convergence')
@mock.patch.object(resource.Resource, 'update_convergence')
def test_check_resource_update_init_action(self, mock_update, mock_create):
self.resource.action = 'INIT'
check_resource.check_resource_update(
self.resource, self.resource.stack.t.id, {}, 'engine-id',
self.stack)
self.assertTrue(mock_create.called)
self.assertFalse(mock_update.called)
@mock.patch.object(resource.Resource, 'create_convergence')
@mock.patch.object(resource.Resource, 'update_convergence')
def test_check_resource_update_create_action(
self, mock_update, mock_create):
self.resource.action = 'CREATE'
check_resource.check_resource_update(
self.resource, self.resource.stack.t.id, {}, 'engine-id',
self.stack)
self.assertFalse(mock_create.called)
self.assertTrue(mock_update.called)
@mock.patch.object(resource.Resource, 'create_convergence')
@mock.patch.object(resource.Resource, 'update_convergence')
def test_check_resource_update_update_action(
self, mock_update, mock_create):
self.resource.action = 'UPDATE'
check_resource.check_resource_update(
self.resource, self.resource.stack.t.id, {}, 'engine-id',
self.stack)
self.assertFalse(mock_create.called)
self.assertTrue(mock_update.called)