Add error handling to background threads

Added error resiliency to the server and client threads, so that exceptions
which occur are logged and the thread continues to run. Additionally, fixed
the logging handler in the server plugin so that stack traces for errors
are added to the output.

Fixes: #5
Change-Id: I6d875a1b3f17f35d29cf7c28297967e0f5045d64

Author: zzzeek

sqlalchemy_collectd/client/tests/test_worker.py

@@ -0,0 +1,88 @@
+import threading
+import unittest
+
+import mock
+
+from .. import worker
+
+
+class WorkerTest(unittest.TestCase):
+    def test_worker_resilency(self):
+        canary = mock.Mock()
+
+        collection_one = ["one", KeyError("random internal error"), "three"]
+        collection_two = ["one", "two", SystemExit()]
+
+        def send(collection, collection_target, now, interval, pid):
+            obj = collection.pop(0)
+            if isinstance(obj, BaseException):
+                raise obj
+            else:
+                canary.send(collection_target, obj)
+
+        the_time = [100]
+
+        mutex = threading.Lock()
+
+        # worker thread will call this at the top of its main loop.
+        def start_loop():
+            # still supporting Python 2, in Py3k only can instead use
+            # nonlocal for "the_time"
+            mutex.acquire()
+            try:
+                return the_time[0]
+            finally:
+                the_time[0] += 5
+                mutex.release()
+
+        with mock.patch.object(
+            worker, "log"
+        ) as mock_logger, mock.patch.object(
+            worker.time, "time", mock.Mock(side_effect=start_loop)
+        ), mock.patch.object(
+            worker.time, "sleep"
+        ):
+            mutex.acquire()
+            try:
+                # this adds the target and also starts the worker thread.
+                # however we have it blocked from doing anything via the
+                # mutex above...
+                worker.add_target(
+                    collection_one, "target one", mock.Mock(send=send)
+                )
+
+                # ...so that we can also add this target and get deterministic
+                # results
+                worker.add_target(
+                    collection_two, "target two", mock.Mock(send=send)
+                )
+            finally:
+                # worker thread is unblocked
+                mutex.release()
+
+            # now wait, it will hit the SystemExit and exit.
+            # if it times out, we failed.
+            worker._WORKER_THREAD.join(1)
+
+        # see that it did what we asked.
+        self.assertEqual(
+            [
+                mock.call.send("target one", "one"),
+                mock.call.send("target two", "one"),
+                mock.call.send("target two", "two"),
+                mock.call.send("target one", "three"),
+            ],
+            canary.mock_calls,
+        )
+
+        self.assertEqual(
+            [
+                mock.call.info("Starting process thread in pid %s", mock.ANY),
+                mock.call.error("error sending stats", exc_info=True),
+                mock.call.info(
+                    "message sender thread caught SystemExit "
+                    "exception, exiting"
+                ),
+            ],
+            mock_logger.mock_calls,
+        )

sqlalchemy_collectd/client/worker.py

@@ -27,19 +27,30 @@ def _process(interval):
     pid = os.getpid()
     log.info("Starting process thread in pid %s", pid)
 
-    while True:
-        now = time.time()
-        for (
-            collection_target,
-            connection,
-            sender,
-            last_called,
-        ) in _collection_targets:
-            if now - last_called[0] > interval:
-                last_called[0] = now
-                sender.send(connection, collection_target, now, interval, pid)
-
-        time.sleep(0.2)
+    try:
+        while True:
+            now = time.time()
+            for (
+                collection_target,
+                connection,
+                sender,
+                last_called,
+            ) in _collection_targets:
+                if now - last_called[0] > interval:
+                    last_called[0] = now
+                    try:
+                        sender.send(
+                            connection, collection_target, now, interval, pid
+                        )
+                    except Exception:
+                        log.error("error sending stats", exc_info=True)
+
+            time.sleep(0.2)
+    except BaseException as be:
+        log.info(
+            "message sender thread caught %s exception, exiting"
+            % type(be).__name__
+        )
 
 
 def add_target(connection, collection_target, sender):

sqlalchemy_collectd/server/listener.py

@@ -1,12 +1,25 @@
+import logging
 import threading
 
+log = logging.getLogger(__name__)
+
 
 def _receive(connection, receiver):
     while True:
-        receiver.receive(connection)
+        try:
+            receiver.receive(connection)
+        except Exception:
+            log.error("message receiver caught an exception", exc_info=True)
+        except BaseException as be:
+            log.info(
+                "message receiver thread caught %s exception, exiting"
+                % type(be).__name__
+            )
+            break
 
 
 def listen(connection, receiver):
+    global listen_thread
     listen_thread = threading.Thread(
         target=_receive, args=(connection, receiver)
     )

sqlalchemy_collectd/server/plugin.py

@@ -27,7 +27,8 @@ class CollectdHandler(logging.Handler):
 
     def emit(self, record):
         fn = self.levels[record.levelno]
-        fn(record.msg % record.args)
+        record.msg = "[sqlalchemy-collectd] " + record.msg
+        fn(self.format(record))
 
 
 def get_config(config):

sqlalchemy_collectd/server/tests/test_listener.py

@@ -0,0 +1,56 @@
+import unittest
+
+import mock
+
+from .. import listener
+
+
+class ListenerTest(unittest.TestCase):
+    def test_receive_resilency(self):
+        canary = mock.Mock()
+
+        collection = [
+            "one",
+            "two",
+            "three",
+            KeyError("random internal error"),
+            "four",
+            SystemExit(),
+            "five",
+        ]
+
+        def receive(connection):
+
+            obj = collection.pop(0)
+            if isinstance(obj, BaseException):
+                raise obj
+            else:
+                canary.receive(obj)
+
+        with mock.patch.object(listener, "log") as mock_logger:
+            listener.listen(mock.Mock(), mock.Mock(receive=receive))
+
+        listener.listen_thread.join(1)
+
+        # call "five" doesn't happen because we should have exited
+        self.assertEqual(
+            [
+                mock.call.receive("one"),
+                mock.call.receive("two"),
+                mock.call.receive("three"),
+                mock.call.receive("four"),
+            ],
+            canary.mock_calls,
+        )
+        self.assertEqual(
+            [
+                mock.call.error(
+                    "message receiver caught an exception", exc_info=True
+                ),
+                mock.call.info(
+                    "message receiver thread caught SystemExit "
+                    "exception, exiting"
+                ),
+            ],
+            mock_logger.mock_calls,
+        )

tox.ini

@@ -29,6 +29,7 @@ deps=
       flake8-builtins
       flake8-docstrings
       flake8-rst-docstrings
+      pydocstyle<4.0.0
       # used by flake8-rst-docstrings
       pygments
 commands = flake8 ./sqlalchemy_collectd/ setup.py

unreleased_changes/5.rst

@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug
+    :tickets: 5
+
+    Added error resiliency to the server and client threads, so that exceptions
+    which occur are logged and the thread continues to run. Additionally, fixed
+    the logging handler in the server plugin so that stack traces for errors
+    are added to the output.