Added lock for log records buffer to avoid a race condition that leads to lost records. Fixes #13 (#18)

Author: hstadler

cmreslogging/handlers.py

@@ -4,7 +4,7 @@
 import logging
 import datetime
 import socket
-from threading import Timer
+from threading import Timer, Lock
 from enum import Enum
 from elasticsearch import helpers as eshelpers
 from elasticsearch import Elasticsearch, RequestsHttpConnection
@@ -197,6 +197,7 @@ def __init__(self,
 
         self._client = None
         self._buffer = []
+        self._buffer_lock = Lock()
         self._timer = None
         self._index_name_func = CMRESHandler._INDEX_FREQUENCY_FUNCION_DICT[self.index_name_frequency]
         self.serializer = CMRESSerializer()
@@ -285,13 +286,16 @@ def flush(self):
 
         if self._buffer:
             try:
+                with self._buffer_lock:
+                    logs_buffer = self._buffer
+                    self._buffer = []
                 actions = (
                     {
                         '_index': self._index_name_func.__func__(self.es_index_name),
                         '_type': self.es_doc_type,
                         '_source': log_record
                     }
-                    for log_record in self._buffer
+                    for log_record in logs_buffer
                 )
                 eshelpers.bulk(
                     client=self.__get_es_client(),
@@ -301,7 +305,6 @@ def flush(self):
             except Exception as exception:
                 if self.raise_on_indexing_exceptions:
                     raise exception
-            self._buffer = []
 
     def close(self):
         """ Flushes the buffer and release any outstanding resource
@@ -327,7 +330,8 @@ def emit(self, record):
             if key not in CMRESHandler.__LOGGING_FILTER_FIELDS:
                 rec[key] = "" if value is None else value
         rec[self.default_timestamp_field_name] = self.__get_es_datetime_str(record.created)
-        self._buffer.append(rec)
+        with self._buffer_lock:
+            self._buffer.append(rec)
 
         if len(self._buffer) >= self.buffer_size:
             self.flush()