Merge pull request #185 from labthings/handle-action-exceptions

Handle action exceptions

If an HTTPException is thrown by an action's thread, before the request has been responded to, propagate the exception in the request handler so that it aborts the response with the right error code.  This means marshmallow validation on action arguments now works.

Author: rwb27

src/labthings/actions/pool.py

@@ -29,15 +29,21 @@ def start(self, thread: ActionThread):
         self.add(thread)
         thread.start()
 
-    def spawn(self, action: str, function, *args, **kwargs):
+    def spawn(self, action: str, function, *args, http_error_lock=None, **kwargs):
         """
 
         :param function:
         :param *args:
         :param **kwargs:
 
         """
-        thread = ActionThread(action, target=function, args=args, kwargs=kwargs)
+        thread = ActionThread(
+            action,
+            target=function,
+            http_error_lock=http_error_lock,
+            args=args,
+            kwargs=kwargs,
+        )
         self.start(thread)
         return thread
 

src/labthings/actions/thread.py

@@ -7,7 +7,7 @@
 from typing import Any, Callable, Dict, Iterable, Optional
 
 from flask import copy_current_request_context, has_request_context, request
-from werkzeug.exceptions import BadRequest
+from werkzeug.exceptions import BadRequest, HTTPException
 
 from ..deque import LockableDeque
 from ..utilities import TimeoutTracker
@@ -22,6 +22,42 @@ class ActionKilledException(SystemExit):
 class ActionThread(threading.Thread):
     """
     A native thread with extra functionality for tracking progress and thread termination.
+
+    Arguments:
+    * `action` is the name of the action that's running
+    * `target`, `name`, `args`, `kwargs` and `daemon` are passed to `threading.Thread`
+      (though the defualt for `daemon` is changed to `True`)
+    * `default_stop_timeout` specifies how long we wait for the `target` function to
+      stop nicely (e.g. by checking the `stopping` Event )
+    * `log_len` gives the number of log entries before we start dumping them
+    * `http_error_lock` allows the calling thread to handle some
+      errors initially.  See below.
+
+    ## Error propagation
+    If the `target` function throws an Exception, by default this will result in:
+    * The thread terminating
+    * The Action's status being set to `error`
+    * The exception appearing in the logs with a traceback
+    * The exception being raised in the background thread.
+    However, `HTTPException` subclasses are used in Flask/Werkzeug web apps to
+    return HTTP status codes indicating specific errors, and so merit being
+    handled differently.
+
+    Normally, when an Action is initiated, the thread handling the HTTP request
+    does not return immediately - it waits for a short period to check whether
+    the Action has completed or returned an error.  If an HTTPError is raised
+    in the Action thread before the initiating thread has sent an HTTP response,
+    we **don't** want to propagate the error here, but instead want to re-raise
+    it in the calling thread.  This will then mean that the HTTP request is
+    answered with the appropriate error code, rather than returning a `201`
+    code, along with a description of the task (showing that it was successfully
+    started, but also showing that it subsequently failed with an error).
+
+    In order to activate this behaviour, we must pass in a `threading.Lock`
+    object.  This lock should already be acquired by the request-handling
+    thread.  If an error occurs, and this lock is acquired, the exception
+    should not be re-raised until the calling thread has had the chance to deal
+    with it.
     """
 
     def __init__(
@@ -34,6 +70,7 @@ def __init__(
         daemon: bool = True,
         default_stop_timeout: int = 5,
         log_len: int = 100,
+        http_error_lock: Optional[threading.Lock] = None,
     ):
         threading.Thread.__init__(
             self,
@@ -56,6 +93,8 @@ def __init__(
         # Event to track if the user has requested stop
         self.stopping: threading.Event = threading.Event()
         self.default_stop_timeout: int = default_stop_timeout
+        # Allow the calling thread to handle HTTP errors for a short time at the start
+        self.http_error_lock = http_error_lock or threading.Lock()
 
         # Make _target, _args, and _kwargs available to the subclass
         self._target: Optional[Callable] = target
@@ -85,6 +124,7 @@ def __init__(
         self._request_time: datetime.datetime = datetime.datetime.now()
         self._start_time: Optional[datetime.datetime] = None  # Task start time
         self._end_time: Optional[datetime.datetime] = None  # Task end time
+        self._exception: Optional[Exception] = None  # Propagate exceptions helpfully
 
         # Public state properties
         self.progress: Optional[int] = None  # Percent progress of the task
@@ -151,6 +191,11 @@ def cancelled(self) -> bool:
         """Alias of `stopped`"""
         return self.stopped
 
+    @property
+    def exception(self) -> Optional[Exception]:
+        """The Exception that caused the action to fail."""
+        return self._exception
+
     def update_progress(self, progress: int):
         """
         Update the progress of the ActionThread.
@@ -214,15 +259,29 @@ def wrapped(*args, **kwargs):
                 # Set state to stopped
                 self._status = "cancelled"
                 self.progress = None
+            except HTTPException as e:
+                self._exception = e
+                # If the lock is acquired elsewhere, assume the error
+                # will be handled there.
+                if self.http_error_lock.acquire(blocking=False):
+                    self.http_error_lock.release()
+                    logging.error(
+                        "An HTTPException occurred in an action thread, but "
+                        "the parent request was no longer waiting for it."
+                    )
+                    logging.error(traceback.format_exc())
+                    raise e
             except Exception as e:  # skipcq: PYL-W0703
+                self._exception = e
                 logging.error(traceback.format_exc())
-                self._return_value = str(e)
-                self._status = "error"
                 raise e
             finally:
                 self._end_time = datetime.datetime.now()
                 logging.getLogger().removeHandler(handler)  # Stop logging this thread
                 # If we don't remove the handler, it's a memory leak.
+                if self._exception:
+                    self._return_value = str(self._exception)
+                    self._status = "error"
 
         return wrapped
 

src/labthings/schema.py

@@ -82,7 +82,11 @@ def preprocess(self, data, **_):
 
 
 class ActionSchema(Schema):
-    """ """
+    """Represents a running or completed Action
+
+    Actions can run in the background, started by one request
+    and subsequently polled for updates.  This schema represents
+    one Action."""
 
     action = fields.String()
     _ID = fields.String(data_key="id")

src/labthings/views/__init__.py

@@ -1,10 +1,12 @@
 import datetime
+import threading
 from collections import OrderedDict
 from typing import Callable, Dict, List, Optional, Set, cast
 
 from flask import request
 from flask.views import MethodView
 from typing_extensions import Protocol
+from werkzeug.exceptions import HTTPException
 from werkzeug.wrappers import Response as ResponseBase
 
 from ..actions.pool import Pool
@@ -215,25 +217,38 @@ def dispatch_request(self, *args, **kwargs):
         pool = (
             current_labthing().actions if current_labthing() else self._emergency_pool
         )
-        # Make a task out of the views `post` method
-        task = pool.spawn(self.endpoint, meth, *args, **kwargs)
-        # Optionally override the threads default_stop_timeout
-        if self.default_stop_timeout is not None:
-            task.default_stop_timeout = self.default_stop_timeout
-
-        # Wait up to 2 second for the action to complete or error
-        try:
-            task.get(block=True, timeout=self.wait_for)
-        except TimeoutError:
-            pass
-
-        # Log the action to the view's deque
-        self._deque.append(task)
+        # We pass in this lock to tell the Action thread that we'll deal
+        # with HTTP errors in this thread
+        error_lock = threading.RLock()
+        with error_lock:
+            # Make a task out of the views `post` method
+            task = pool.spawn(
+                self.endpoint, meth, *args, http_error_lock=error_lock, **kwargs
+            )
+            # Optionally override the threads default_stop_timeout
+            if self.default_stop_timeout is not None:
+                task.default_stop_timeout = self.default_stop_timeout
+
+            # Log the action to the view's deque
+            self._deque.append(task)
+
+            # Wait up to 2 second for the action to complete or error
+            try:
+                task.get(block=True, timeout=self.wait_for)
+            except TimeoutError:
+                pass
 
         # If the action returns quickly, and returns a valid Response, return it as-is
         if task.output and isinstance(task.output, ResponseBase):
             return self.represent_response((task.output, 200))
 
+        # If the action fails quickly with an HTTPException, propagate it.
+        # This allows us to handle validation errors nicely.
+        # Similarly, calling Flask's `abort(404)` will work during the
+        # timeout period, as it uses the same mechanism.
+        if task.exception and isinstance(task.exception, HTTPException):
+            raise task.exception
+
         return self.represent_response((ActionSchema().dump(task), 201))
 
 

tests/conftest.py

@@ -1,11 +1,12 @@
 import json
 import os
+import time
 
 import jsonschema
 import pytest
 from apispec import APISpec
 from apispec.ext.marshmallow import MarshmallowPlugin
-from flask import Flask
+from flask import Flask, abort
 from flask.testing import FlaskClient
 from flask.views import MethodView
 from marshmallow import validate
@@ -188,7 +189,7 @@ class TestAction(ActionView):
         def post(self):
             return "POST"
 
-    thing.add_view(TestAction, "TestAction")
+    thing.add_view(TestAction, "/TestAction")
 
     class TestProperty(PropertyView):
         schema = {"count": fields.Integer()}
@@ -199,7 +200,7 @@ def get(self):
         def post(self, args):
             pass
 
-    thing.add_view(TestProperty, "TestProperty")
+    thing.add_view(TestProperty, "/TestProperty")
 
     class TestFieldProperty(PropertyView):
         schema = fields.String(validate=validate.OneOf(["one", "two"]))
@@ -210,7 +211,35 @@ def get(self):
         def post(self, args):
             pass
 
-    thing.add_view(TestFieldProperty, "TestFieldProperty")
+    thing.add_view(TestFieldProperty, "/TestFieldProperty")
+
+    class FailAction(ActionView):
+        wait_for = 0.1
+
+        def post(self):
+            raise Exception("This action is meant to fail with an Exception")
+
+    thing.add_view(FailAction, "/FailAction")
+
+    class AbortAction(ActionView):
+        wait_for = 0.1
+        args = {"abort_after": fields.Number()}
+
+        def post(self, args):
+            if args.get("abort_after", 0) > 0:
+                time.sleep(args["abort_after"])
+            abort(418, "I'm a teapot! This action should abort with an HTTP code 418")
+
+    thing.add_view(AbortAction, "/AbortAction")
+
+    class ActionWithValidation(ActionView):
+        wait_for = 0.1
+        args = {"test_arg": fields.String(validate=validate.OneOf(["one", "two"]))}
+
+        def post(self, args):
+            return True
+
+    thing.add_view(ActionWithValidation, "/ActionWithValidation")
 
     return thing
 

tests/test_action_api.py

@@ -0,0 +1,74 @@
+import json
+import logging
+import time
+
+import pytest
+
+from labthings import LabThing
+from labthings.views import ActionView
+
+
+@pytest.mark.filterwarnings("ignore:Exception in thread")
+def test_action_exception_handling(thing_with_some_views, client):
+    """Check errors in an Action are handled correctly
+
+
+
+    `/FieldProperty` has a validation constraint - it
+    should return a "bad response" error if invoked with
+    anything other than
+    """
+    # `/FailAction` raises an `Exception`.
+    # This ought to return a 201 code representing the
+    # action that was successfully started - but should
+    # show that it failed through the "status" field.
+
+    # This is correct for the current (24/7/2021) behaviour
+    # but may want to change for the next version, e.g.
+    # returning a 500 code.  For further discussion...
+    r = client.post("/FailAction")
+    assert r.status_code == 201
+    action = r.get_json()
+    assert action["status"] == "error"
+
+
+def test_action_abort(thing_with_some_views, client):
+    """Check HTTPExceptions result in error codes.
+
+    Subclasses of HTTPError should result in a non-200 return code, not
+    just failures.  This covers Marshmallow validation (400) and
+    use of `abort()`.
+    """
+    # `/AbortAction` should return a 418 error code
+    r = client.post("/AbortAction")
+    assert r.status_code == 418
+
+
+@pytest.mark.filterwarnings("ignore:Exception in thread")
+def test_action_abort_late(thing_with_some_views, client, caplog):
+    """Check HTTPExceptions raised late are just regular errors."""
+    caplog.set_level(logging.ERROR)
+    caplog.clear()
+    r = client.post("/AbortAction", data=json.dumps({"abort_after": 0.2}))
+    assert r.status_code == 201  # Should have started OK
+    time.sleep(0.3)
+    # Now check the status - should be error
+    r2 = client.get(r.get_json()["links"]["self"]["href"])
+    assert r2.get_json()["status"] == "error"
+    # Check it was logged as well
+    error_was_raised = False
+    for r in caplog.records:
+        if r.levelname == "ERROR" and "HTTPException" in r.message:
+            error_was_raised = True
+    assert error_was_raised
+
+
+def test_action_validate(thing_with_some_views, client):
+    """Validation errors should result in 422 return codes."""
+    # `/ActionWithValidation` should fail with a 400 error
+    # if `test_arg` is not either `one` or `two`
+    r = client.post("/ActionWithValidation", data=json.dumps({"test_arg": "one"}))
+    assert r.status_code in [200, 201]
+    assert r.get_json()["status"] == "completed"
+    r = client.post("/ActionWithValidation", data=json.dumps({"test_arg": "three"}))
+    assert r.status_code in [422]

tests/test_labthing_exceptions.py

@@ -65,7 +65,7 @@ def test_blank_exception(app):
     e = Exception()
     e.message = None
 
-    # Test a 404 HTTPException
+    # Test an empty Exception
     response = error_handler.std_handler(e)
 
     response_json = json.dumps(response[0])