Handle OS::Mistral::Workflow resource replacement properly

OS::Mistral::Workflow resource creates a mistral workflow with
a unique name (resource_id). We replace FAILED resources by
default and replace wont work in this case as it will try to
use the same workflow name for the replacement resouce, if the
'name' property is provided.

If the workflow does not exist/deleted using mistral api directly,
it would create a replacement resource, but it would delete the
workflow when cleaning up the old resource. So we would endup
with a replacement resource without any backing workflow.

This adds a new resource attribute ``always_replace_on_check_failed``
and overrides needs_replace_failed() for OS::Mistral::Workflow.

Task: 38855
Change-Id: Ia0812b88cae363dfa25ccd907ecbe8b86f5b1a23
(cherry picked from commit 9e80518b900a1f14a80ee05ddeef4b250433febd)

3 files changed, 87 insertions, 21 deletions

diff --git a/heat/engine/resource.py b/heat/engine/resource.py
index c383f0abf..b6827cca1 100644
--- a/heat/engine/resource.py
+++ b/heat/engine/resource.py
@@ -159,6 +159,9 @@ class Resource(status.ResourceStatus):
     # a signal to this resource
     signal_needs_metadata_updates = True
 
+    # Whether the resource is always replaced when CHECK_FAILED
+    always_replace_on_check_failed = True
+
     def __new__(cls, name, definition, stack):
         """Create a new Resource of the appropriate class for its type."""
 
@@ -1387,7 +1390,9 @@ class Resource(status.ResourceStatus):
                       prev_resource, check_init_complete=True):
         if self.status == self.FAILED:
             # always replace when a resource is in CHECK_FAILED
-            if self.action == self.CHECK or self.needs_replace_failed():
+            if ((self.action == self.CHECK and
+                 self.always_replace_on_check_failed) or (
+                     self.needs_replace_failed())):
                 raise UpdateReplace(self)
 
         if self.state == (self.DELETE, self.COMPLETE):
diff --git a/heat/engine/resources/openstack/mistral/workflow.py b/heat/engine/resources/openstack/mistral/workflow.py
index 3c9ea23bc..8794b23d7 100644
--- a/heat/engine/resources/openstack/mistral/workflow.py
+++ b/heat/engine/resources/openstack/mistral/workflow.py
@@ -46,6 +46,8 @@ class Workflow(signal_responder.SignalResponder,
 
     entity = 'workflows'
 
+    always_replace_on_check_failed = False
+
     PROPERTIES = (
         NAME, TYPE, DESCRIPTION, INPUT, OUTPUT, TASKS, PARAMS,
         TASK_DEFAULTS, USE_REQUEST_BODY_AS_INPUT, TAGS
@@ -596,6 +598,19 @@ class Workflow(signal_responder.SignalResponder,
             executions.extend(self.data().get(self.EXECUTIONS).split(','))
         self.data_set(self.EXECUTIONS, ','.join(executions))
 
+    def needs_replace_failed(self):
+        if self.resource_id is None:
+            return True
+
+        if self.properties[self.NAME] is None:
+            return True
+
+        with self.client_plugin().ignore_not_found:
+            self.client().workflows.get(self.resource_id)
+            return False
+        self.resource_id_set(None)
+        return True
+
     def handle_update(self, json_snippet, tmpl_diff, prop_diff):
         if prop_diff:
             props = json_snippet.properties(self.properties_schema,
diff --git a/heat/tests/openstack/mistral/test_workflow.py b/heat/tests/openstack/mistral/test_workflow.py
index 641014e07..face6859a 100644
--- a/heat/tests/openstack/mistral/test_workflow.py
+++ b/heat/tests/openstack/mistral/test_workflow.py
@@ -15,6 +15,7 @@ import mock
 import six
 import yaml
 
+from mistralclient.api import base as mistral_base
 from mistralclient.api.v2 import executions
 from oslo_serialization import jsonutils
 
@@ -302,6 +303,21 @@ resources:
             result: <% $.hello %>
 """
 
+workflow_template_update_replace_failed = """
+heat_template_version: 2013-05-23
+resources:
+  workflow:
+    type: OS::Mistral::Workflow
+    properties:
+      name: hello_action
+      type: direct
+      tasks:
+        - name: hello
+          action: std.echo output='Good Morning!'
+          publish:
+            result: <% $.hello %>
+"""
+
 workflow_template_update = """
 heat_template_version: 2013-05-23
 resources:
@@ -373,12 +389,6 @@ class TestMistralWorkflow(common.HeatTestCase):
     def setUp(self):
         super(TestMistralWorkflow, self).setUp()
         self.ctx = utils.dummy_context()
-        tmpl = template_format.parse(workflow_template)
-        self.stack = utils.parse_stack(tmpl, stack_name='test_stack')
-
-        resource_defns = self.stack.t.resource_definitions(self.stack)
-        self.rsrc_defn = resource_defns['workflow']
-
         self.mistral = mock.Mock()
         self.patchobject(workflow.Workflow, 'client',
                          return_value=self.mistral)
@@ -400,15 +410,20 @@ class TestMistralWorkflow(common.HeatTestCase):
         for patch in self.patches:
             patch.stop()
 
-    def _create_resource(self, name, snippet, stack):
-        wf = workflow.Workflow(name, snippet, stack)
+    def _create_resource(self, name, template=workflow_template):
+        tmpl = template_format.parse(template)
+        self.stack = utils.parse_stack(tmpl, stack_name='test_stack')
+
+        resource_defns = self.stack.t.resource_definitions(self.stack)
+        rsrc_defn = resource_defns['workflow']
+        wf = workflow.Workflow(name, rsrc_defn, self.stack)
         self.mistral.workflows.create.return_value = [
             FakeWorkflow('test_stack-workflow-b5fiekfci3yc')]
         scheduler.TaskRunner(wf.create)()
         return wf
 
     def test_create(self):
-        wf = self._create_resource('workflow', self.rsrc_defn, self.stack)
+        wf = self._create_resource('workflow')
         expected_state = (wf.CREATE, wf.COMPLETE)
         self.assertEqual(expected_state, wf.state)
         self.assertEqual('test_stack-workflow-b5fiekfci3yc', wf.resource_id)
@@ -460,7 +475,7 @@ class TestMistralWorkflow(common.HeatTestCase):
                 break
 
     def test_attributes(self):
-        wf = self._create_resource('workflow', self.rsrc_defn, self.stack)
+        wf = self._create_resource('workflow')
         self.mistral.workflows.get.return_value = (
             FakeWorkflow('test_stack-workflow-b5fiekfci3yc'))
         self.assertEqual({'name': 'test_stack-workflow-b5fiekfci3yc',
@@ -521,7 +536,7 @@ class TestMistralWorkflow(common.HeatTestCase):
                       six.text_type(exc))
 
     def test_update_replace(self):
-        wf = self._create_resource('workflow', self.rsrc_defn, self.stack)
+        wf = self._create_resource('workflow')
 
         t = template_format.parse(workflow_template_update_replace)
         rsrc_defns = template.Template(t).resource_definitions(self.stack)
@@ -538,8 +553,7 @@ class TestMistralWorkflow(common.HeatTestCase):
         self.assertEqual(msg, six.text_type(err))
 
     def test_update(self):
-        wf = self._create_resource('workflow', self.rsrc_defn,
-                                   self.stack)
+        wf = self._create_resource('workflow')
         t = template_format.parse(workflow_template_update)
         rsrc_defns = template.Template(t).resource_definitions(self.stack)
         new_wf = rsrc_defns['workflow']
@@ -550,8 +564,7 @@ class TestMistralWorkflow(common.HeatTestCase):
         self.assertEqual((wf.UPDATE, wf.COMPLETE), wf.state)
 
     def test_update_input(self):
-        wf = self._create_resource('workflow', self.rsrc_defn,
-                                   self.stack)
+        wf = self._create_resource('workflow')
         t = template_format.parse(workflow_template)
         t['resources']['workflow']['properties']['input'] = {'foo': 'bar'}
         rsrc_defns = template.Template(t).resource_definitions(self.stack)
@@ -563,8 +576,7 @@ class TestMistralWorkflow(common.HeatTestCase):
         self.assertEqual((wf.UPDATE, wf.COMPLETE), wf.state)
 
     def test_update_failed(self):
-        wf = self._create_resource('workflow', self.rsrc_defn,
-                                   self.stack)
+        wf = self._create_resource('workflow')
         t = template_format.parse(workflow_template_update)
         rsrc_defns = template.Template(t).resource_definitions(self.stack)
         new_wf = rsrc_defns['workflow']
@@ -573,8 +585,42 @@ class TestMistralWorkflow(common.HeatTestCase):
                           scheduler.TaskRunner(wf.update, new_wf))
         self.assertEqual((wf.UPDATE, wf.FAILED), wf.state)
 
+    def test_update_failed_no_replace(self):
+        wf = self._create_resource('workflow',
+                                   workflow_template_update_replace)
+        t = template_format.parse(workflow_template_update_replace_failed)
+        rsrc_defns = template.Template(t).resource_definitions(self.stack)
+        new_wf = rsrc_defns['workflow']
+        self.mistral.workflows.get.return_value = (
+            FakeWorkflow('test_stack-workflow-b5fiekfci3yc'))
+        self.mistral.workflows.update.side_effect = [
+            Exception('boom!'),
+            [FakeWorkflow('test_stack-workflow-b5fiekfci3yc')]]
+        self.assertRaises(exception.ResourceFailure,
+                          scheduler.TaskRunner(wf.update, new_wf))
+        self.assertEqual((wf.UPDATE, wf.FAILED), wf.state)
+        scheduler.TaskRunner(wf.update, new_wf)()
+        self.assertTrue(self.mistral.workflows.update.called)
+        self.assertEqual((wf.UPDATE, wf.COMPLETE), wf.state)
+
+    def test_update_failed_replace_not_found(self):
+        wf = self._create_resource('workflow',
+                                   workflow_template_update_replace)
+        t = template_format.parse(workflow_template_update_replace_failed)
+        rsrc_defns = template.Template(t).resource_definitions(self.stack)
+        new_wf = rsrc_defns['workflow']
+        self.mistral.workflows.update.side_effect = Exception('boom!')
+        self.assertRaises(exception.ResourceFailure,
+                          scheduler.TaskRunner(wf.update, new_wf))
+        self.assertEqual((wf.UPDATE, wf.FAILED), wf.state)
+        self.mistral.workflows.get.side_effect = [
+            mistral_base.APIException(error_code=404)]
+        self.assertRaises(resource.UpdateReplace,
+                          scheduler.TaskRunner(wf.update,
+                                               new_wf))
+
     def test_delete_super_call_successful(self):
-        wf = self._create_resource('workflow', self.rsrc_defn, self.stack)
+        wf = self._create_resource('workflow')
 
         scheduler.TaskRunner(wf.delete)()
         self.assertEqual((wf.DELETE, wf.COMPLETE), wf.state)
@@ -582,7 +628,7 @@ class TestMistralWorkflow(common.HeatTestCase):
         self.assertEqual(1, self.mistral.workflows.delete.call_count)
 
     def test_delete_executions_successful(self):
-        wf = self._create_resource('workflow', self.rsrc_defn, self.stack)
+        wf = self._create_resource('workflow')
 
         self.mistral.executuions.delete.return_value = None
         wf._data = {'executions': '1234,5678'}
@@ -594,7 +640,7 @@ class TestMistralWorkflow(common.HeatTestCase):
         data_delete.assert_called_once_with('executions')
 
     def test_delete_executions_not_found(self):
-        wf = self._create_resource('workflow', self.rsrc_defn, self.stack)
+        wf = self._create_resource('workflow')
 
         self.mistral.executuions.delete.side_effect = [
             self.mistral.mistral_base.APIException(error_code=404),