Merge pull request #1330 from PrefectHQ/add-sh-log

Add error log for exceptions raised in state handlers with tests

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
 

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)

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

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 = []

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):