<commit_before># 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

import eventlet.queue
import functools

from oslo_log import log as logging
from oslo_utils import excutils

from heat.common import exception
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 api as rpc_api
from heat.rpc import listener_client

LOG = logging.getLogger(__name__)


class CancelOperation(BaseException):
    """Exception to cancel an in-progress operation on a resource.

    This exception is raised when operations on a resource are cancelled.
    """
    def __init__(self):
        return super(CancelOperation, self).__init__('user triggered cancel')


class CheckResource(object):

    def __init__(self,
                 engine_id,
                 rpc_client,
                 thread_group_mgr,
                 msg_queue,
                 input_data):
        self.engine_id = engine_id
        self._rpc_client = rpc_client
        self.thread_group_mgr = thread_group_mgr
        self.msg_queue = msg_queue
        self.input_data = input_data

    def _try_steal_engine_lock(self, cnxt, resource_id):
        rs_obj = resource_objects.Resource.get_obj(cnxt,
                                                   resource_id,
                                                   fields=('engine_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("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,
                                 stack.RESTORE)):
            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.
            self._retrigger_new_traversal(cnxt, stack.current_traversal,
                                          is_update,
                                          stack.id, rsrc_id)

    def _retrigger_new_traversal(self, cnxt, current_traversal, is_update,
                                 stack_id, rsrc_id):
            latest_stack = parser.Stack.load(cnxt, stack_id=stack_id,
                                             force_reload=True)
            if current_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 _handle_resource_replacement(self, cnxt,
                                     current_traversal, new_tmpl_id,
                                     rsrc, stack, adopt_stack_data):
        """Create a replacement resource and trigger a check on it."""
        try:
            new_res_id = rsrc.make_replacement(new_tmpl_id)
        except exception.UpdateInProgress:
            LOG.info("No replacement created - "
                     "resource already locked by new traversal")
            return
        if new_res_id is None:
            LOG.info("No replacement created - "
                     "new traversal already in progress")
            self._retrigger_new_traversal(cnxt, current_traversal, True,
                                          stack.id, rsrc.id)
            return
        LOG.info("Replacing resource with new id %s", new_res_id)
        rpc_data = sync_point.serialize_input_data(self.input_data)
        self._rpc_client.check_resource(cnxt,
                                        new_res_id,
                                        current_traversal,
                                        rpc_data, True,
                                        adopt_stack_data)

    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, self.msg_queue)
                except resource.UpdateReplace:
                    self._handle_resource_replacement(cnxt, current_traversal,
                                                      tmpl.id,
                                                      rsrc, stack,
                                                      adopt_stack_data)
                    return False

            else:
                check_resource_cleanup(rsrc, tmpl.id, resource_data,
                                       self.engine_id,
                                       stack.time_remaining(), self.msg_queue)

            return True
        except exception.UpdateInProgress:
            if self._try_steal_engine_lock(cnxt, rsrc.id):
                rpc_data = sync_point.serialize_input_data(self.input_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:
            action = ex.action or rsrc.action
            reason = 'Resource %s failed: %s' % (action,
                                                 six.text_type(ex))
            self._handle_resource_failure(cnxt, is_update, rsrc.id,
                                          stack, reason)
        except scheduler.Timeout:
            self._handle_resource_failure(cnxt, is_update, rsrc.id,
                                          stack, u'Timed out')
        except CancelOperation:
            pass

        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-trigger received for update in latest traversal, first
            # check if update key is available in graph.
            # if No, then latest traversal is waiting for delete.
            if (resource_id, is_update) not in graph:
                key = (resource_id, not is_update)
        else:
            # When re-trigger received for delete in latest traversal, first
            # check if update key is available in graph,
            # if yes, then latest traversal is waiting for update.
            if (resource_id, True) in graph:
                # not is_update evaluates to True below, which means update
                key = (resource_id, not is_update)
        LOG.info('Re-trigger resource: (%(key1)s, %(key2)s)',
                 {'key1': key[0], 'key2': 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:<commit_msg>Retrigger new traversals after resource cancellation<commit_after># 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

import eventlet.queue
import functools

from oslo_log import log as logging
from oslo_utils import excutils

from heat.common import exception
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 api as rpc_api
from heat.rpc import listener_client

LOG = logging.getLogger(__name__)


class CancelOperation(BaseException):
    """Exception to cancel an in-progress operation on a resource.

    This exception is raised when operations on a resource are cancelled.
    """
    def __init__(self):
        return super(CancelOperation, self).__init__('user triggered cancel')


class CheckResource(object):

    def __init__(self,
                 engine_id,
                 rpc_client,
                 thread_group_mgr,
                 msg_queue,
                 input_data):
        self.engine_id = engine_id
        self._rpc_client = rpc_client
        self.thread_group_mgr = thread_group_mgr
        self.msg_queue = msg_queue
        self.input_data = input_data

    def _try_steal_engine_lock(self, cnxt, resource_id):
        rs_obj = resource_objects.Resource.get_obj(cnxt,
                                                   resource_id,
                                                   fields=('engine_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("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,
                                 stack.RESTORE)):
            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.
            self._retrigger_new_traversal(cnxt, stack.current_traversal,
                                          is_update,
                                          stack.id, rsrc_id)

    def _retrigger_new_traversal(self, cnxt, current_traversal, is_update,
                                 stack_id, rsrc_id):
            latest_stack = parser.Stack.load(cnxt, stack_id=stack_id,
                                             force_reload=True)
            if current_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 _handle_resource_replacement(self, cnxt,
                                     current_traversal, new_tmpl_id,
                                     rsrc, stack, adopt_stack_data):
        """Create a replacement resource and trigger a check on it."""
        try:
            new_res_id = rsrc.make_replacement(new_tmpl_id)
        except exception.UpdateInProgress:
            LOG.info("No replacement created - "
                     "resource already locked by new traversal")
            return
        if new_res_id is None:
            LOG.info("No replacement created - "
                     "new traversal already in progress")
            self._retrigger_new_traversal(cnxt, current_traversal, True,
                                          stack.id, rsrc.id)
            return
        LOG.info("Replacing resource with new id %s", new_res_id)
        rpc_data = sync_point.serialize_input_data(self.input_data)
        self._rpc_client.check_resource(cnxt,
                                        new_res_id,
                                        current_traversal,
                                        rpc_data, True,
                                        adopt_stack_data)

    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, self.msg_queue)
                except resource.UpdateReplace:
                    self._handle_resource_replacement(cnxt, current_traversal,
                                                      tmpl.id,
                                                      rsrc, stack,
                                                      adopt_stack_data)
                    return False

            else:
                check_resource_cleanup(rsrc, tmpl.id, resource_data,
                                       self.engine_id,
                                       stack.time_remaining(), self.msg_queue)

            return True
        except exception.UpdateInProgress:
            if self._try_steal_engine_lock(cnxt, rsrc.id):
                rpc_data = sync_point.serialize_input_data(self.input_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:
            action = ex.action or rsrc.action
            reason = 'Resource %s failed: %s' % (action,
                                                 six.text_type(ex))
            self._handle_resource_failure(cnxt, is_update, rsrc.id,
                                          stack, reason)
        except scheduler.Timeout:
            self._handle_resource_failure(cnxt, is_update, rsrc.id,
                                          stack, u'Timed out')
        except CancelOperation as ex:
            # Stack is already marked FAILED, so we just need to retrigger
            # in case a new traversal has started and is waiting on us.
            self._retrigger_new_traversal(cnxt, current_traversal, is_update,
                                          stack.id, rsrc.id)

        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-trigger received for update in latest traversal, first
            # check if update key is available in graph.
            # if No, then latest traversal is waiting for delete.
            if (resource_id, is_update) not in graph:
                key = (resource_id, not is_update)
        else:
            # When re-trigger received for delete in latest traversal, first
            # check if update key is available in graph,
            # if yes, then latest traversal is waiting for update.
            if (resource_id, True) in graph:
                # not is_update evaluates to True below, which means update
                key = (resource_id, not is_update)
        LOG.info('Re-trigger resource: (%(key1)s, %(key2)s)',
                 {'key1': key[0], 'key2': 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:
{'fraction_non_alphanumeric': 0.0634185303514377, 'fraction_numerical': 0.001597444089456869, 'mean_word_length': 1.5486497489482969, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 108, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3490689', 'n_tokens_mistral': 4642, 'n_tokens_neox': 4498, 'n_words': 1279}