Implement two-step logging (#1080)

A few times we observed that Patroni HA loop was blocked for a few minutes due to not being able to write logs to stderr. This is a very rare condition which we hit so far only on k8s. This commit makes Patroni resilient to such kind of problems. All log messages first are written into the in-memory queue and later they are asynchronously flushed into the stderr or file from a separate thread.

The maximum queue size is configurable and the default value is 1000. This should be enough to keep more than one hour of log messages with default settings and when Patroni cluster operates normally (without big issues).

In case if we hit the maximum size of the queue further logs will be discarded until the queue size will be reduced. The number of discarded messages will be reported into the log later.

In addition to that, the number of non-flushed and discarded messages (if there are any), will be reported via Patroni REST API as:
```json
"logger_queue_size": X,
"logger_records_lost": Y`
```

docs/ENVIRONMENT.rst

@@ -14,6 +14,7 @@ Global/Universal
 -  **PATRONI\_LOG\_LEVEL**: sets the general logging level. Default value is **INFO** (see `the docs for Python logging <https://docs.python.org/3.6/library/logging.html#levels>`_)
 -  **PATRONI\_LOG\_FORMAT**: sets the log formatting string. Default value is **%(asctime)s %(levelname)s: %(message)s** (see `the LogRecord attributes <https://docs.python.org/3.6/library/logging.html#logrecord-attributes>`_)
 -  **PATRONI\_LOG\_DATEFORMAT**: sets the datetime formatting string. (see the `formatTime() documentation <https://docs.python.org/3.6/library/logging.html#logging.Formatter.formatTime>`_)
+-  **PATRONI\_LOG\_MAX\_QUEUE\_SIZE**: Patroni is using two-step logging. Log records are written into the in-memory queue and there is a separate thread which pulls them from the queue and writes to stderr or file. The maximum size of the internal queue is limited by default by **1000** records, which is enough to keep logs for the past 1h20m.
 -  **PATRONI\_LOG\_DIR**: Directory to write application logs to. The directory must exist and be writable by the user executing Patroni. If you set this env variable, the application will retain 4 25MB logs by default. You can tune those retention values with `PATRONI_LOG_FILE_NUM` and `PATRONI_LOG_FILE_SIZE` (see below).
 -  **PATRONI\_LOG\_FILE\_NUM**: The number of application logs to retain.
 -  **PATRONI\_LOG\_FILE\_SIZE**: Size of patroni.log file (in bytes) that triggers a log rolling.

docs/SETTINGS.rst

@@ -15,6 +15,7 @@ Log
 -  **level**: sets the general logging level. Default value is **INFO** (see `the docs for Python logging <https://docs.python.org/3.6/library/logging.html#levels>`_)
 -  **format**: sets the log formatting string. Default value is **%(asctime)s %(levelname)s: %(message)s** (see `the LogRecord attributes <https://docs.python.org/3.6/library/logging.html#logrecord-attributes>`_)
 -  **dateformat**: sets the datetime formatting string. (see the `formatTime() documentation <https://docs.python.org/3.6/library/logging.html#logging.Formatter.formatTime>`_)
+-  **max\_queue\_size**: Patroni is using two-step logging. Log records are written into the in-memory queue and there is a separate thread which pulls them from the queue and writes to stderr or file. The maximum size of the internal queue is limited by default by **1000** records, which is enough to keep logs for the past 1h20m.
 -  **dir**: Directory to write application logs to. The directory must exist and be writable by the user executing Patroni. If you set this value, the application will retain 4 25MB logs by default. You can tune those retention values with `file_num` and `file_size` (see below).
 -  **file\_num**: The number of application logs to retain.
 -  **file\_size**: Size of patroni.log file (in bytes) that triggers a log rolling.

patroni/__init__.py

@@ -151,6 +151,7 @@ class Patroni(object):
         except Exception:
             logger.exception('Exception during RestApi.shutdown')
         self.ha.shutdown()
+        self.logger.shutdown()
 
 
 def patroni_main():
@@ -161,7 +162,6 @@ def patroni_main():
         pass
     finally:
         patroni.shutdown()
-        logging.shutdown()
 
 
 def fatal(string, *args):

patroni/api.py

@@ -75,6 +75,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
             response['watchdog_failed'] = True
         if patroni.ha.is_paused():
             response['pause'] = True
+        qsize = patroni.logger.queue_size
+        if qsize > patroni.logger.NORMAL_LOG_QUEUE_SIZE:
+            response['logger_queue_size'] = qsize
+            lost = patroni.logger.records_lost
+            if lost:
+                response['logger_records_lost'] = lost
         self._write_json_response(status_code, response)
 
     def do_GET(self, write_status_code_only=False):

patroni/config.py

@@ -232,7 +232,8 @@ class Config(object):
 
         _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile'])
         _set_section_values('postgresql', ['listen', 'connect_address', 'config_dir', 'data_dir', 'pgpass', 'bin_dir'])
-        _set_section_values('log', ['level', 'format', 'dateformat', 'dir', 'file_size', 'file_num', 'loggers'])
+        _set_section_values('log', ['level', 'format', 'dateformat', 'max_queue_size',
+                                    'dir', 'file_size', 'file_num', 'loggers'])
 
         def _parse_dict(value):
             if not value.strip().startswith('{'):

patroni/log.py

@@ -1,66 +1,158 @@
 import logging
 import os
+import sys
 
 from copy import deepcopy
 from logging.handlers import RotatingFileHandler
 from patroni.utils import deep_compare
+from six.moves.queue import Queue, Full
+from threading import Lock, Thread
+
+_LOGGER = logging.getLogger(__name__)
 
 
-class PatroniLogger(object):
+class QueueHandler(logging.Handler):
+
+    def __init__(self):
+        logging.Handler.__init__(self)
+        self.queue = Queue()
+        self._records_lost = 0
+
+    def _put_record(self, record):
+        self.format(record)
+        record.msg = record.message
+        record.args = None
+        record.exc_info = None
+        self.queue.put_nowait(record)
+
+    def _try_to_report_lost_records(self):
+        if self._records_lost:
+            try:
+                record = _LOGGER.makeRecord(_LOGGER.name, logging.WARNING, __file__, 0,
+                                            'QueueHandler has lost %s log records',
+                                            (self._records_lost,), None, 'emit')
+                self._put_record(record)
+                self._records_lost = 0
+            except Exception:
+                pass
+
+    def emit(self, record):
+        try:
+            self._put_record(record)
+            self._try_to_report_lost_records()
+        except Exception:
+            self._records_lost += 1
+
+    @property
+    def records_lost(self):
+        return self._records_lost
+
+
+class PatroniLogger(Thread):
 
     DEFAULT_LEVEL = 'INFO'
     DEFAULT_FORMAT = '%(asctime)s %(levelname)s: %(message)s'
 
+    NORMAL_LOG_QUEUE_SIZE = 2  # When everything goes normal Patroni writes only 2 messages per HA loop
+    DEFAULT_MAX_QUEUE_SIZE = 1000
+    LOGGING_BROKEN_EXIT_CODE = 5
+
     def __init__(self):
-        self.root_logger = logging.getLogger()
-        self.config = None
-        self.handler = None
+        super(PatroniLogger, self).__init__()
+        self.daemon = True
+        self._queue_handler = QueueHandler()
+        self._root_logger = logging.getLogger()
+        self._root_logger.addHandler(self._queue_handler)
+        self._config = None
+        self._log_handler = None
+        self._old_handlers = []
+        self._log_handler_lock = Lock()
         self.reload_config({'level': 'DEBUG'})
+        self.start()
 
     def update_loggers(self):
-        loggers = deepcopy(self.config.get('loggers') or {})
-        for name, logger in self.root_logger.manager.loggerDict.items():
+        loggers = deepcopy(self._config.get('loggers') or {})
+        for name, logger in self._root_logger.manager.loggerDict.items():
             if not isinstance(logger, logging.PlaceHolder):
                 level = loggers.pop(name, logging.NOTSET)
                 logger.setLevel(level)
 
         for name, level in loggers.items():
-            logger = self.root_logger.manager.getLogger(name)
+            logger = self._root_logger.manager.getLogger(name)
             logger.setLevel(level)
 
     def reload_config(self, config):
-        if self.config is None or not deep_compare(self.config, config):
-            self.root_logger.setLevel(config.get('level', PatroniLogger.DEFAULT_LEVEL))
+        if self._config is None or not deep_compare(self._config, config):
+            with self._queue_handler.queue.mutex:
+                self._queue_handler.queue.maxsize = config.get('max_queue_size', self.DEFAULT_MAX_QUEUE_SIZE)
 
-            add_handler = None
+            self._root_logger.setLevel(config.get('level', PatroniLogger.DEFAULT_LEVEL))
+
+            new_handler = None
             if 'dir' in config:
-                if not isinstance(self.handler, RotatingFileHandler):
-                    add_handler = RotatingFileHandler(os.path.join(config['dir'], __name__))
-                handler = add_handler or self.handler
+                if not isinstance(self._log_handler, RotatingFileHandler):
+                    new_handler = RotatingFileHandler(os.path.join(config['dir'], __name__))
+                handler = new_handler or self._log_handler
                 handler.maxBytes = int(config.get('file_size', 25000000))
                 handler.backupCount = int(config.get('file_num', 4))
             else:
-                if self.handler is None or isinstance(self.handler, RotatingFileHandler):
-                    add_handler = logging.StreamHandler()
-                handler = add_handler or self.handler
+                if self._log_handler is None or isinstance(self._log_handler, RotatingFileHandler):
+                    new_handler = logging.StreamHandler()
+                handler = new_handler or self._log_handler
 
-            oldlogformat = (self.config or {}).get('format', PatroniLogger.DEFAULT_FORMAT)
+            oldlogformat = (self._config or {}).get('format', PatroniLogger.DEFAULT_FORMAT)
             logformat = config.get('format', PatroniLogger.DEFAULT_FORMAT)
 
-            olddateformat = (self.config or {}).get('dateformat') or None
+            olddateformat = (self._config or {}).get('dateformat') or None
             dateformat = config.get('dateformat') or None  # Convert empty string to `None`
 
-            if oldlogformat != logformat or olddateformat != dateformat or add_handler:
+            if oldlogformat != logformat or olddateformat != dateformat or new_handler:
                 handler.setFormatter(logging.Formatter(logformat, dateformat))
 
-            if add_handler:
-                self.root_logger.addHandler(add_handler)
+            if new_handler:
+                with self._log_handler_lock:
+                    if self._log_handler:
+                        self._old_handlers.append(self._log_handler)
+                    self._log_handler = new_handler
 
-                if self.handler is not None:
-                    self.root_logger.removeHandler(self.handler)
-                    self.handler.close()
-
-                self.handler = add_handler
-
-            self.config = config.copy()
+            self._config = config.copy()
             self.update_loggers()
+
+    def _close_old_handlers(self):
+        while True:
+            with self._log_handler_lock:
+                if not self._old_handlers:
+                    break
+                handler = self._old_handlers.pop()
+            try:
+                handler.close()
+            except Exception:
+                _LOGGER.exception('Failed to close the old log handler %s', handler)
+
+    def run(self):
+        while True:
+            self._close_old_handlers()
+
+            record = self._queue_handler.queue.get(True)
+            if record is None:
+                break
+
+            self._log_handler.handle(record)
+            self._queue_handler.queue.task_done()
+
+    def shutdown(self):
+        try:
+            self._queue_handler.queue.put_nowait(None)
+        except Full:  # Queue is full.
+            # It seems that logging is not working, exiting with non-standard exit-code is the best we can do.
+            sys.exit(self.LOGGING_BROKEN_EXIT_CODE)
+        self.join()
+        logging.shutdown()
+
+    @property
+    def queue_size(self):
+        return self._queue_handler.queue.qsize()
+
+    @property
+    def records_lost(self):
+        return self._queue_handler.records_lost

tests/test_api.py

@@ -100,12 +100,20 @@ class MockHa(object):
         return False
 
 
+class MockLogger(object):
+
+    NORMAL_LOG_QUEUE_SIZE = 2
+    queue_size = 3
+    records_lost = 1
+
+
 class MockPatroni(object):
 
     ha = MockHa()
     config = Mock()
     postgresql = ha.state_handler
     dcs = Mock()
+    logger = MockLogger()
     tags = {}
     version = '0.00'
     noloadbalance = PropertyMock(return_value=False)

tests/test_log.py

@@ -7,6 +7,7 @@ import yaml
 from mock import Mock, patch
 from patroni.config import Config
 from patroni.log import PatroniLogger
+from six.moves.queue import Queue, Full
 
 
 class TestPatroniLogger(unittest.TestCase):
@@ -18,9 +19,11 @@ class TestPatroniLogger(unittest.TestCase):
         logging.getLogger().handlers[:] = self._handlers
 
     @patch('logging.FileHandler._open', Mock())
+    @patch('logging.Handler.close', Mock(side_effect=Exception))
     def test_patroni_logger(self):
         config = {
             'log': {
+                'max_queue_size': 5,
                 'dir': 'foo',
                 'file_size': 4096,
                 'file_num': 5,
@@ -36,8 +39,20 @@ class TestPatroniLogger(unittest.TestCase):
         patroni_config = Config()
         logger.reload_config(patroni_config['log'])
 
-        self.assertEqual(logger.handler.maxBytes, config['log']['file_size'])
-        self.assertEqual(logger.handler.backupCount, config['log']['file_num'])
+        with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)):
+            logging.error('test')
+
+        self.assertEqual(logger._log_handler.maxBytes, config['log']['file_size'])
+        self.assertEqual(logger._log_handler.backupCount, config['log']['file_num'])
 
         config['log'].pop('dir')
         logger.reload_config(config['log'])
+        with patch.object(logging.Logger, 'makeRecord',
+                          Mock(side_effect=[logging.LogRecord('', logging.INFO, '', 0, '', (), None), Exception])):
+            logging.error('test')
+        logging.error('test')
+        with patch.object(Queue, 'put_nowait', Mock(side_effect=Full)):
+            self.assertRaises(SystemExit, logger.shutdown)
+        self.assertRaises(Exception, logger.shutdown)
+        self.assertLessEqual(logger.queue_size, 2)  # "Failed to close the old log handler" could be still in the queue
+        self.assertEqual(logger.records_lost, 0)

tests/test_patroni.py

@@ -1,5 +1,6 @@
 import etcd
 import logging
+import os
 import signal
 import sys
 import time
@@ -14,6 +15,7 @@ from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
 from patroni import Patroni, main as _main, patroni_main, check_psycopg2
 from six.moves import BaseHTTPServer, builtins
+from threading import Thread
 
 from . import psycopg2_connect, SleepException
 from .test_etcd import etcd_read, etcd_write
@@ -37,6 +39,7 @@ class MockFrozenImporter(object):
 @patch.object(AsyncExecutor, 'run', Mock())
 @patch.object(etcd.Client, 'write', etcd_write)
 @patch.object(etcd.Client, 'read', etcd_read)
+@patch.object(Thread, 'start', Mock())
 class TestPatroni(unittest.TestCase):
 
     @patch('pkgutil.get_importer', Mock(return_value=MockFrozenImporter()))
@@ -51,6 +54,7 @@ class TestPatroni(unittest.TestCase):
         with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             sys.argv = ['patroni.py', 'postgres0.yml']
+            os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0'
             self.p = Patroni()
 
     def tearDown(self):
@@ -65,6 +69,7 @@ class TestPatroni(unittest.TestCase):
     @patch('time.sleep', Mock(side_effect=SleepException))
     @patch.object(etcd.Client, 'delete', Mock())
     @patch.object(Client, 'machines')
+    @patch.object(Thread, 'join', Mock())
     def test_patroni_patroni_main(self, mock_machines):
         with patch('subprocess.call', Mock(return_value=1)):
             sys.argv = ['patroni.py', 'postgres0.yml']