Protect prepare_update_replace() with resource lock

When we consolidated resource locking with resource state transitions in
a7376f7494, prepare_update_replace() moved
outside of the locked section, so that other update operations could
potentially conflict with it. This change ensures that we call it while the
lock is still held (if we have transitioned the resource to
UPDATE_IN_PROGRESS, and thus acquired the lock), or that we acquire the
lock before calling it (if UpdateReplace is raised before attempting to
update the resource).

To avoid unnecessary locking and unlocking, we only take the lock if the
Resource plugin has overridden the default handler method (either
restore_prev_rsrc() or prepare_for_replace()), since the defaults do
nothing.

Change-Id: Ie32836ed643e440143bde8b83aeb4d6159acd034
Closes-Bug: #1727127

heat/engine/resource.py

@@ -1481,16 +1481,34 @@ class Resource(status.ResourceStatus):
                             "error: %s", ex)
         return after_props, before_props
 
-    def _prepare_update_replace(self, action):
-        try:
+    def _prepare_update_replace_handler(self, action):
+        """Return the handler method for preparing to replace a resource.
+
+        This may be either restore_prev_rsrc() (in the case of a legacy
+        rollback) or, more typically, prepare_for_replace().
+
+        If the plugin has not overridden the method, then None is returned in
+        place of the default method (which is empty anyway).
+        """
         if (self.stack.action == 'ROLLBACK' and
                 self.stack.status == 'IN_PROGRESS' and
                 not self.stack.convergence):
             # handle case, when it's rollback and we should restore
             # old resource
-                self.restore_prev_rsrc()
+            if self.restore_prev_rsrc != Resource.restore_prev_rsrc:
+                return self.restore_prev_rsrc
         else:
-                self.prepare_for_replace()
+            if self.prepare_for_replace != Resource.prepare_for_replace:
+                return self.prepare_for_replace
+        return None
+
+    def _prepare_update_replace(self, action):
+        handler = self._prepare_update_replace_handler(action)
+        if handler is None:
+            return
+
+        try:
+            handler()
         except Exception as e:
             # if any exception happen, we should set the resource to
             # FAILED, then raise ResourceFailure
@@ -1549,6 +1567,16 @@ class Resource(status.ResourceStatus):
                 needs_update = self._needs_update(after, before,
                                                   after_props, before_props,
                                                   prev_resource)
+        except UpdateReplace:
+            with excutils.save_and_reraise_exception():
+                if self._prepare_update_replace_handler(action) is not None:
+                    with self.lock(self._calling_engine_id):
+                        self._prepare_update_replace(action)
+        except exception.ResourceActionRestricted as ae:
+            failure = exception.ResourceFailure(ae, self, action)
+            self._add_event(action, self.FAILED, six.text_type(ae))
+            raise failure
+
         if not needs_update:
             if update_templ_func is not None:
                 update_templ_func(persist=True)
@@ -1576,18 +1604,22 @@ class Resource(status.ResourceStatus):
 
         with self._action_recorder(action, UpdateReplace):
             after_props.validate()
-
+            self.properties = before_props
             tmpl_diff = self.update_template_diff(after.freeze(), before)
+
+            try:
                 if tmpl_diff and self.needs_replace_with_tmpl_diff(tmpl_diff):
                     raise UpdateReplace(self)
 
                 prop_diff = self.update_template_diff_properties(after_props,
                                                                  before_props)
-                self.properties = before_props
-
                 yield self.action_handler_task(action,
                                                args=[after, tmpl_diff,
                                                      prop_diff])
+            except UpdateReplace:
+                with excutils.save_and_reraise_exception():
+                    self._prepare_update_replace(action)
+
             self.t = after
             self.reparse()
             self._update_stored_properties()
@@ -1595,16 +1627,6 @@ class Resource(status.ResourceStatus):
                 # template/requires will be persisted by _action_recorder()
                 update_templ_func(persist=False)
 
-        except exception.ResourceActionRestricted as ae:
-            # catch all ResourceActionRestricted exceptions
-            failure = exception.ResourceFailure(ae, self, action)
-            self._add_event(action, self.FAILED, six.text_type(ae))
-            raise failure
-        except UpdateReplace:
-            # catch all UpdateReplace exceptions
-            self._prepare_update_replace(action)
-            raise
-
         yield self._break_if_required(
             self.UPDATE, environment.HOOK_POST_UPDATE)
 
@@ -2066,14 +2088,17 @@ class Resource(status.ResourceStatus):
     def lock(self, engine_id):
         self._calling_engine_id = engine_id
         try:
+            if engine_id is not None:
                 self._store_with_lock({}, self.LOCK_ACQUIRE)
             yield
         except exception.UpdateInProgress:
             raise
         except BaseException:
             with excutils.save_and_reraise_exception():
+                if engine_id is not None:
                     self._store_with_lock({}, self.LOCK_RELEASE)
         else:
+            if engine_id is not None:
                 self._store_with_lock({}, self.LOCK_RELEASE)
 
     def _resolve_any_attribute(self, attr):

heat/tests/test_resource.py

@@ -2409,7 +2409,7 @@ class ResourceTest(common.HeatTestCase):
         self.assertEqual(stack.t.id, res.current_template_id)
         # ensure that requires was not updated
         self.assertItemsEqual([2], res.requires)
-        self._assert_resource_lock(res.id, None, None)
+        self._assert_resource_lock(res.id, None, 2)
 
     def test_convergence_update_replace_rollback(self):
         rsrc_def = rsrc_defn.ResourceDefinition('test_res',