Use regular queue, use flush thread per process, delay worker init (#11)

* Use regular queue, use flush thread per process

* Small changes, fix tests

* Changes, tests

Author: curusarn

logtail/handler.py

@@ -32,30 +32,33 @@ def __init__(self,
         self.source_token = source_token
         self.host = host
         self.context = context
-        self.pipe = multiprocessing.JoinableQueue(maxsize=buffer_capacity)
+        self.pipe = queue.Queue(maxsize=buffer_capacity)
         self.uploader = Uploader(self.source_token, self.host)
         self.drop_extra_events = drop_extra_events
         self.include_extra_attributes = include_extra_attributes
         self.buffer_capacity = buffer_capacity
         self.flush_interval = flush_interval
         self.raise_exceptions = raise_exceptions
         self.dropcount = 0
+        # Do not initialize the flush thread yet because it causes issues on Render.
+        self.flush_thread = None
+
+    def ensure_flush_thread_alive(self):
+        if self.flush_thread and self.flush_thread.is_alive():
+            return
+
         self.flush_thread = FlushWorker(
             self.uploader,
             self.pipe,
             self.buffer_capacity,
             self.flush_interval
         )
-        if self._is_main_process():
-            self.flush_thread.start()
-
-    def _is_main_process(self):
-        return multiprocessing.current_process()._parent_pid == None
+        self.flush_thread.start()
 
     def emit(self, record):
         try:
-            if self._is_main_process() and not self.flush_thread.is_alive():
-                self.flush_thread.start()
+            self.ensure_flush_thread_alive()
+
             message = self.format(record)
             frame = create_frame(record, message, self.context, include_extra_attributes=self.include_extra_attributes)
             try:

tests/test_handler.py

@@ -28,32 +28,43 @@ def test_handler_creates_pipe_from_args(self, MockWorker):
             buffer_capacity=buffer_capacity,
             flush_interval=flush_interval
         )
-        self.assertEqual(handler.pipe._maxsize, buffer_capacity)
+        self.assertTrue(handler.pipe.empty())
 
     @mock.patch('logtail.handler.FlushWorker')
-    def test_handler_creates_and_starts_worker_from_args(self, MockWorker):
+    def test_handler_creates_and_starts_worker_from_args_after_first_log(self, MockWorker):
         buffer_capacity = 9
         flush_interval = 9
         handler = LogtailHandler(source_token=self.source_token, buffer_capacity=buffer_capacity, flush_interval=flush_interval)
+
+        self.assertFalse(MockWorker.called)
+
+        logger = logging.getLogger(__name__)
+        logger.handlers = []
+        logger.addHandler(handler)
+        logger.critical('hello')
+
         MockWorker.assert_called_with(
             handler.uploader,
             handler.pipe,
             buffer_capacity,
             flush_interval
         )
-        self.assertTrue(handler.flush_thread.start.called)
+        self.assertEqual(handler.flush_thread.start.call_count, 1)
 
     @mock.patch('logtail.handler.FlushWorker')
     def test_emit_starts_thread_if_not_alive(self, MockWorker):
         handler = LogtailHandler(source_token=self.source_token)
-        self.assertTrue(handler.flush_thread.start.call_count, 1)
-        handler.flush_thread.is_alive = mock.Mock(return_value=False)
 
         logger = logging.getLogger(__name__)
         logger.handlers = []
         logger.addHandler(handler)
         logger.critical('hello')
 
+        self.assertEqual(handler.flush_thread.start.call_count, 1)
+        handler.flush_thread.is_alive = mock.Mock(return_value=False)
+
+        logger.critical('hello')
+
         self.assertEqual(handler.flush_thread.start.call_count, 2)
 
     @mock.patch('logtail.handler.FlushWorker')