[Fixes #270] Make BackgroundThread more resilient to exceptions (#297)

Author: Olivier Cervello

opencensus/common/transports/async.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import atexit
+import logging
 import threading
 import time
 
@@ -100,7 +101,15 @@ def _thread_main(self):
                     data.extend(item)
 
             if data:
-                self.exporter.emit(data)
+                try:
+                    self.exporter.emit(data)
+                except Exception:
+                    logging.exception(
+                        '%s failed to emit data.'
+                        'Dropping %s objects from queue.',
+                        self.exporter.__class__.__name__,
+                        len(data))
+                    pass
 
             for _ in range(len(items)):
                 self._queue.task_done()

opencensus/trace/exporters/transports/background_thread.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import atexit
+import logging
 import threading
 import time
 
@@ -101,7 +102,15 @@ def _thread_main(self):
                     span_datas.extend(item)
 
             if span_datas:
-                self.exporter.emit(span_datas)
+                try:
+                    self.exporter.emit(span_datas)
+                except Exception:
+                    logging.exception(
+                        '%s failed to emit spans.'
+                        'Dropping %s spans from queue.',
+                        self.exporter.__class__.__name__,
+                        len(span_datas))
+                    pass
 
             for _ in range(len(items)):
                 self._queue.task_done()

tests/unit/common/transports/test_async.py

@@ -200,6 +200,45 @@ def emit(self, span):
         # trace2 should be left in the queue because worker is terminated.
         self.assertEqual(worker._queue.qsize(), 1)
 
+    @mock.patch('logging.exception')
+    def test__thread_main_alive_on_emit_failed(self, mock):
+
+        class Exporter(object):
+            def __init__(self):
+                self.exported = []
+
+            def emit(self, span):
+                if len(self.exported) < 2:
+                    self.exported.extend(span)
+                else:
+                    raise Exception("This exporter is broken !")
+
+        exporter = Exporter()
+        worker = async._Worker(exporter, max_batch_size=2)
+
+        span_data0 = [mock.Mock()]
+        span_data1 = [mock.Mock()]
+        span_data2 = [mock.Mock()]
+
+        worker.enqueue(span_data0)
+        worker.enqueue(span_data1)
+        worker.enqueue(span_data2)
+        worker.enqueue(async._WORKER_TERMINATOR)
+
+        worker._thread_main()
+
+        # Span 2 should throw an exception, only span 0 and 1 are left
+        self.assertEqual(exporter.exported, span_data0 + span_data1)
+
+        # Logging exception should have been called on the exporter exception
+        expected = '%s failed to emit data.Dropping %s objects from queue.'
+        mock.assert_called_with(expected, 'Exporter', 1)
+
+        # Nothing should be left in the queue because worker is terminated
+        # and the data was dropped.
+        self.assertEqual(worker._queue.qsize(), 0)
+
+
     def test_flush(self):
         from six.moves import queue
 

tests/unit/trace/exporters/transports/test_background_thread.py

@@ -200,6 +200,44 @@ def emit(self, span):
         # trace2 should be left in the queue because worker is terminated.
         self.assertEqual(worker._queue.qsize(), 1)
 
+    @mock.patch('logging.exception')
+    def test__thread_main_alive_on_emit_failed(self, mock):
+
+        class Exporter(object):
+            def __init__(self):
+                self.exported = []
+
+            def emit(self, span):
+                if len(self.exported) < 2:
+                    self.exported.extend(span)
+                else:
+                    raise Exception("This exporter is broken !")
+
+        exporter = Exporter()
+        worker = background_thread._Worker(exporter, max_batch_size=2)
+
+        span_data0 = [mock.Mock()]
+        span_data1 = [mock.Mock()]
+        span_data2 = [mock.Mock()]
+
+        worker.enqueue(span_data0)
+        worker.enqueue(span_data1)
+        worker.enqueue(span_data2)
+        worker.enqueue(background_thread._WORKER_TERMINATOR)
+
+        worker._thread_main()
+
+        # Span 2 should throw an exception, only span 0 and 1 are left
+        self.assertEqual(exporter.exported, span_data0 + span_data1)
+
+        # Logging exception should have been called on the exporter exception
+        expected = '%s failed to emit spans.Dropping %s spans from queue.'
+        mock.assert_called_with(expected, 'Exporter', 1)
+
+        # Nothing should be left in the queue because worker is terminated
+        # and the data was dropped.
+        self.assertEqual(worker._queue.qsize(), 0)
+
     def test_flush(self):
         from six.moves import queue