Merge pull request #253 from crungehottman/logging-thread-retries

Add fatal LoggingThread error backout strategy

kobo/worker/logger.py

@@ -2,6 +2,7 @@
 
 import threading
 import time
+import os
 
 import six
 
@@ -31,6 +32,7 @@ def __init__(self, hub, task_id, *args, **kwargs):
         self._running = True
         self._send_time = 0
         self._send_data = b""
+        self._timeout = int(os.environ.get("KOBO_LOGGING_THREAD_TIMEOUT", 600))
 
     def read_queue(self):
         out = self._queue.get_nowait()
@@ -68,15 +70,26 @@ def run(self):
                 self._send_time = now
                 self._send_data = b""
             except Exception:
-                # Log all caught exceptions.
+                # Any exception other than an XML-RPC fault may be fatal. It is
+                # possible that we've encountered a retryable error, such as a
+                # temporary network disruption between worker and hub. Attempt
+                # to retry for a bit.
+                if now - self._send_time <= self._timeout:
+                    continue
+
+                # If the timemout has been exceeded, we can assume we've
+                # encountered a non-temporary, fatal exception.
+                #
+                # Since upload_task_log is apparently not working, we can't get
+                # this into the task logs, but it should at least be possible
+                # for this to get into the worker's local log file.
                 if self._logger:
                     msg = "\n".join([
-                        "Exception in LoggingThread:",
+                        "Fatal error in LoggingThread",
                         kobo.tback.Traceback().get_traceback(),
                     ])
-                    self._logger.log_error(msg)
-
-                continue
+                    self._logger.log_critical(msg)
+                raise
 
     def write(self, data):
         """Add data to the queue and set the event for sending queue content."""

tests/test_logger.py

@@ -56,6 +56,33 @@ def test_upload_task_log_after_some_time(self):
         self.assertFalse(thread.is_alive())
         self.assertFalse(thread._running)
 
+    # Following test intentionally kills a thread with an exception.
+    @pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning")
+    def test_logs_on_fatal_error(self):
+        # Set up a logger whose output we'll be able to inspect.
+        logs = StringIO()
+        logger = logging.getLogger('TestLoggingThread')
+        logger.addHandler(logging.StreamHandler(logs))
+        kobo_logger = LoggingBase(logger)
+
+        mock_hub = Mock()
+        mock_hub.upload_task_log.side_effect = RuntimeError("Simulated error")
+
+        thread = LoggingThread(mock_hub, 9999, logger=kobo_logger)
+        thread.daemon = True
+        thread.start()
+
+        thread.write('This is a log message!')
+        # Since we set up a fatal error, we expect the thread to die soon
+        # despite not calling stop().
+        thread.join(10.0)
+        self.assertFalse(thread.is_alive())
+
+        # Before dying, it should have written something useful to the logs.
+        captured = logs.getvalue()
+        self.assertIn('Fatal error in LoggingThread', captured)
+        self.assertIn('RuntimeError: Simulated error', captured)
+
     def test_logs_during_temporary_outage(self):
         # Messages written to the logging thread during a temporary
         # outage should be uploaded (and not discarded) after the hub