fix: Remove Thread/event to fix Django regression

Author: Stephen Hawkins

logdna/logdna.py

@@ -76,24 +76,17 @@ def __init__(self, key, options={}):
         self.setLevel(logging.DEBUG)
         self.lock = threading.RLock()
 
-        # Start the flusher
-        self.flusher_stopped = threading.Event()
-        self.flusher = threading.Thread(
-            target=self.flush_timer_worker, daemon=True)
-        self.flusher.start()
-
-    def flush_timer_worker(self):
-        while not self.flusher_stopped.wait(self.flush_interval_secs):
-            try:
-                self.flush()
-            except Exception as e:
-                self.internalLogger.exception(
-                    f'Error in flush_timer_worker: {e}')
+        self.flusher = None
+
+    def start_flusher(self):
+        if not self.flusher:
+            self.flusher = threading.Timer(
+                self.flush_interval_secs, self.flush)
+            self.flusher.start()
 
     def close_flusher(self):
         if self.flusher:
-            self.flusher_stopped.set()
-            self.flusher.join()
+            self.flusher.cancel()
             self.flusher = None
 
     def buffer_log(self, message):
@@ -119,7 +112,10 @@ def buffer_log_sync(self, message):
                         self.buf_retention_limit)
 
                 if self.buf_size >= self.flush_limit:
+                    self.close_flusher()
                     self.flush()
+                else:
+                    self.start_flusher()
             except Exception as e:
                 self.internalLogger.exception(f'Error in buffer_log_sync: {e}')
             finally:
@@ -143,13 +139,15 @@ def try_lock_and_do_flush_request(self, should_block=False):
         local_buf = []
         if self.lock.acquire(blocking=should_block):
             if not self.buf:
+                self.close_flusher()
                 self.lock.release()
                 return
 
-            if self.buf:
-                local_buf = self.buf.copy()
-                self.buf.clear()
-                self.buf_size = 0
+            local_buf = self.buf.copy()
+            self.buf.clear()
+            self.buf_size = 0
+            if local_buf:
+                self.close_flusher()
             self.lock.release()
 
         if local_buf:
@@ -301,10 +299,10 @@ def emit(self, record):
             'line': msg,
             'level': record['levelname'] or self.loglevel,
             'app': self.app or record['module'],
-            'env': self.env
+            'env': self.env,
+            'meta': {}
         }
 
-        message['meta'] = {}
         for key in self.custom_fields:
             if key in record:
                 if isinstance(record[key], tuple):
@@ -325,6 +323,9 @@ def emit(self, record):
         self.buffer_log(message)
 
     def close(self):
+        # Close the flusher
+        self.close_flusher()
+
         # First gracefully shut down any threads that are still attempting
         # to add log messages to the buffer. This ensures that we don't lose
         # any log messages that are in the process of being added to the
@@ -333,10 +334,6 @@ def close(self):
             self.worker_thread_pool.shutdown(wait=True)
             self.worker_thread_pool = None
 
-        # Now that we've shut down the worker threads, we can safely close
-        # the flusher thread.
-        self.close_flusher()
-
         # Manually force a flush of any remaining log messages in the buffer.
         # We block here to ensure that the flush completes prior to the
         # application exiting and because the probability of this