From 7e25f74879c64147d1fd13fd708fa950155b49ba Mon Sep 17 00:00:00 2001
From: Chris White <chris@prefect.io>
Date: Wed, 7 Aug 2019 12:15:06 -0700
Subject: [PATCH] Add error log for exceptions raised in state handlers with
 tests

---
 CHANGELOG.md                                 |  1 +
 src/prefect/engine/cloud/task_runner.py      |  2 +-
 src/prefect/engine/runner.py                 |  6 +++---
 tests/engine/cloud/test_cloud_task_runner.py |  8 +++++++-
 tests/engine/test_task_runner.py             | 15 +++++++++++++++
 5 files changed, 27 insertions(+), 5 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d25bbbe8fcf0..846e54c5c543 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 - Allow passing of custom headers in `Client` calls - [#1255](https://github.com/PrefectHQ/prefect/pull/1255)
 - Autogenerate informative names and tags for Docker images built for Flow storage - [#1237](https://github.com/PrefectHQ/prefect/issues/1237)
 - Allow mixed-case configuration keys (environment variables are interpolated as lowercase) - [#1288](https://github.com/PrefectHQ/prefect/issues/1288)
+- Ensure state handler errors are logged informatively - [#1326](https://github.com/PrefectHQ/prefect/issues/1326)
 
 ### Task Library
 
diff --git a/src/prefect/engine/cloud/task_runner.py b/src/prefect/engine/cloud/task_runner.py
index 0b374fc457ba..34b88ef975e0 100644
--- a/src/prefect/engine/cloud/task_runner.py
+++ b/src/prefect/engine/cloud/task_runner.py
@@ -78,7 +78,7 @@ def call_runner_target_handlers(self, old_state: State, new_state: State) -> Sta
             )
         except Exception as exc:
             msg = "Exception raised while calling state handlers: {}".format(repr(exc))
-            self.logger.debug(msg)
+            self.logger.error(msg)
             if raise_on_exception:
                 raise exc
             new_state = Failed(msg, result=exc)
diff --git a/src/prefect/engine/runner.py b/src/prefect/engine/runner.py
index 2cbc82ad0d1e..aa903d9698bb 100644
--- a/src/prefect/engine/runner.py
+++ b/src/prefect/engine/runner.py
@@ -178,7 +178,7 @@ def handle_state_change(self, old_state: State, new_state: State) -> State:
         except Exception as exc:
             if raise_on_exception:
                 raise
-            raise ENDRUN(
-                Failed("Exception raised while calling state handlers.", result=exc)
-            )
+            msg = "Exception raised while calling state handlers: {}".format(repr(exc))
+            self.logger.error(msg)
+            raise ENDRUN(Failed(msg, result=exc))
         return new_state
diff --git a/tests/engine/cloud/test_cloud_task_runner.py b/tests/engine/cloud/test_cloud_task_runner.py
index 71967e989444..1c6185f0725d 100644
--- a/tests/engine/cloud/test_cloud_task_runner.py
+++ b/tests/engine/cloud/test_cloud_task_runner.py
@@ -585,7 +585,7 @@ def add(x, y):
         assert states[2].cached_inputs == dict(x=x, y=y)
 
 
-def test_state_handler_failures_are_handled_appropriately(client):
+def test_state_handler_failures_are_handled_appropriately(client, caplog):
     def bad(*args, **kwargs):
         raise SyntaxError("Syntax Errors are nice because they're so unique")
 
@@ -604,6 +604,12 @@ def do_nothing():
     assert states[1].is_failed()
     assert isinstance(states[1].result, SyntaxError)
 
+    error_logs = [r.message for r in caplog.records if r.levelname == "ERROR"]
+    assert len(error_logs) == 1
+    assert "SyntaxError" in error_logs[0]
+    assert "unique" in error_logs[0]
+    assert "state handler" in error_logs[0]
+
 
 def test_task_runner_performs_retries_for_short_delays(client):
     global_list = []
diff --git a/tests/engine/test_task_runner.py b/tests/engine/test_task_runner.py
index 46aa10c5f4d4..a6b5fbdb6c10 100644
--- a/tests/engine/test_task_runner.py
+++ b/tests/engine/test_task_runner.py
@@ -1357,6 +1357,21 @@ def test_task_handler_that_doesnt_return_state_or_none(self):
             with prefect.utilities.debug.raise_on_exception():
                 TaskRunner(task=task).run()
 
+    def test_task_handler_errors_are_logged(self, caplog):
+        def handler(*args, **kwargs):
+            raise SyntaxError("oops")
+
+        task = Task(state_handlers=[handler])
+        state = TaskRunner(task=task).run()
+
+        assert state.is_failed()
+
+        error_logs = [r.message for r in caplog.records if r.levelname == "ERROR"]
+        assert len(error_logs) == 1
+        assert "SyntaxError" in error_logs[0]
+        assert "oops" in error_logs[0]
+        assert "state handler" in error_logs[0]
+
 
 class TestTaskRunnerStateHandlers:
     def test_task_runner_handlers_are_called(self):