Re-raise DB errors when evaluating expressions

* See details in the bug description

Change-Id: Id8c00c5d1196b72c08aa1d518b25a08290868693
Closes-Bug: #1729582

mistral/expressions/jinja_expression.py

@@ -17,6 +17,7 @@ import re
 import jinja2
 from jinja2 import parser as jinja_parse
 from jinja2.sandbox import SandboxedEnvironment
+from oslo_db import exception as db_exc
 from oslo_log import log as logging
 import six
 
@@ -77,6 +78,24 @@ class JinjaEvaluator(Evaluator):
             # to access it is to try and cast it to string.
             str(result)
         except Exception as e:
+            # NOTE(rakhmerov): if we hit a database error then we need to
+            # re-raise the initial exception so that upper layers had a
+            # chance to handle it properly (e.g. in case of DB deadlock
+            # the operations needs to retry. Essentially, such situation
+            # indicates a problem with DB rather than with the expression
+            # syntax or values.
+            if isinstance(e, db_exc.DBError):
+                LOG.error(
+                    "Failed to evaluate Jinja expression due to a database"
+                    " error, re-raising initial exception [expression=%s,"
+                    " error=%s, data=%s]",
+                    expression,
+                    str(e),
+                    data_context
+                )
+
+                raise e
+
             raise exc.JinjaEvaluationException(
                 "Can not evaluate Jinja expression [expression=%s, error=%s"
                 ", data=%s]" % (expression, str(e), data_context)

mistral/expressions/yaql_expression.py

@@ -17,6 +17,7 @@
 import inspect
 import re
 
+from oslo_db import exception as db_exc
 from oslo_log import log as logging
 import six
 from yaql.language import exceptions as yaql_exc
@@ -49,6 +50,24 @@ class YAQLEvaluator(Evaluator):
                 context=expression_utils.get_yaql_context(data_context)
             )
         except Exception as e:
+            # NOTE(rakhmerov): if we hit a database error then we need to
+            # re-raise the initial exception so that upper layers had a
+            # chance to handle it properly (e.g. in case of DB deadlock
+            # the operations needs to retry. Essentially, such situation
+            # indicates a problem with DB rather than with the expression
+            # syntax or values.
+            if isinstance(e, db_exc.DBError):
+                LOG.error(
+                    "Failed to evaluate YAQL expression due to a database"
+                    " error, re-raising initial exception [expression=%s,"
+                    " error=%s, data=%s]",
+                    expression,
+                    str(e),
+                    data_context
+                )
+
+                raise e
+
             raise exc.YaqlEvaluationException(
                 "Can not evaluate YAQL expression [expression=%s, error=%s"
                 ", data=%s]" % (expression, str(e), data_context)

mistral/tests/unit/engine/test_error_handling.py

@@ -12,13 +12,16 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
+import mock
 from oslo_config import cfg
+from oslo_db import exception as db_exc
 
 from mistral.db.v2 import api as db_api
 from mistral import exceptions as exc
 from mistral.services import workbooks as wb_service
 from mistral.services import workflows as wf_service
 from mistral.tests.unit.engine import base
+from mistral.utils import expression_utils
 from mistral.workflow import states
 from mistral_lib import actions as actions_base
 
@@ -741,3 +744,83 @@ class ErrorHandlingEngineTest(base.EngineTestCase):
 
         self.assertIn("UnicodeDecodeError: utf", wf_ex.state_info)
         self.assertIn("UnicodeDecodeError: utf", task_ex.state_info)
+
+    @mock.patch(
+        'mistral.utils.expression_utils.get_yaql_context',
+        mock.MagicMock(
+            side_effect=[
+                db_exc.DBDeadlock(),  # Emulating DB deadlock
+                expression_utils.get_yaql_context({})  # Successful run
+            ]
+        )
+    )
+    def test_db_error_in_yaql_expression(self):
+        # This test just checks that the workflow completes successfully
+        # even if a DB deadlock occurs during YAQL expression evaluation.
+        # The engine in this case should should just retry the transactional
+        # method.
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task1:
+              action: std.echo output="Hello"
+              publish:
+                my_var: <% 1 + 1 %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        wf_ex = self.engine.start_workflow('wf', '', {})
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            self.assertEqual(1, len(wf_ex.task_executions))
+
+            task_ex = wf_ex.task_executions[0]
+
+            self.assertDictEqual({'my_var': 2}, task_ex.published)
+
+    @mock.patch(
+        'mistral.utils.expression_utils.get_jinja_context',
+        mock.MagicMock(
+            side_effect=[
+                db_exc.DBDeadlock(),  # Emulating DB deadlock
+                expression_utils.get_jinja_context({})  # Successful run
+            ]
+        )
+    )
+    def test_db_error_in_jinja_expression(self):
+        # This test just checks that the workflow completes successfully
+        # even if a DB deadlock occurs during Jinja expression evaluation.
+        # The engine in this case should should just retry the transactional
+        # method.
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task1:
+              action: std.echo output="Hello"
+              publish:
+                my_var: "{{ 1 + 1 }}"
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        wf_ex = self.engine.start_workflow('wf', '', {})
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            self.assertEqual(1, len(wf_ex.task_executions))
+
+            task_ex = wf_ex.task_executions[0]
+
+            self.assertDictEqual({'my_var': 2}, task_ex.published)