Explicitly start logger Thread (#1186)

The PatroniLogger object is instantiated in the Patroni constructor and down the road there might be a fatal error causing Patroni process to exit, but live thread prevents the normal shutdown.
In order to mitigate the issue and don't loose ability to use the logging infrastructure we will switch to QueueLogger only when the thread was explicitly started from the Patroni.run() method.

Continuation of https://github.com/zalando/patroni/pull/1178

patroni/__init__.py

@@ -114,6 +114,7 @@ class Patroni(object):
 
     def run(self):
         self.api.start()
+        self.logger.start()
         self.next_run = time.time()
 
         while not self.received_sigterm:

patroni/log.py

@@ -48,6 +48,16 @@ class QueueHandler(logging.Handler):
         return self._records_lost
 
 
+class ProxyHandler(logging.Handler):
+
+    def __init__(self, patroni_logger):
+        logging.Handler.__init__(self)
+        self.patroni_logger = patroni_logger
+
+    def emit(self, record):
+        self.patroni_logger.log_handler.handle(record)
+
+
 class PatroniLogger(Thread):
 
     DEFAULT_LEVEL = 'INFO'
@@ -61,13 +71,16 @@ class PatroniLogger(Thread):
         super(PatroniLogger, self).__init__()
         self._queue_handler = QueueHandler()
         self._root_logger = logging.getLogger()
-        self._root_logger.addHandler(self._queue_handler)
         self._config = None
-        self._log_handler = None
+        self.log_handler = None
+        self.log_handler_lock = Lock()
         self._old_handlers = []
-        self._log_handler_lock = Lock()
         self.reload_config({'level': 'DEBUG'})
-        self.start()
+        # We will switch to the QueueHandler only when thread was started.
+        # This is necessary to protect from the cases when Patroni constructor
+        # failed and PatroniLogger thread remain running and prevent shutdown.
+        self._proxy_handler = ProxyHandler(self)
+        self._root_logger.addHandler(self._proxy_handler)
 
     def update_loggers(self):
         loggers = deepcopy(self._config.get('loggers') or {})
@@ -89,15 +102,15 @@ class PatroniLogger(Thread):
 
             new_handler = None
             if 'dir' in config:
-                if not isinstance(self._log_handler, RotatingFileHandler):
+                if not isinstance(self.log_handler, RotatingFileHandler):
                     new_handler = RotatingFileHandler(os.path.join(config['dir'], __name__))
-                handler = new_handler or self._log_handler
+                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._log_handler is None or isinstance(self._log_handler, RotatingFileHandler):
+                if self.log_handler is None or isinstance(self.log_handler, RotatingFileHandler):
                     new_handler = logging.StreamHandler()
-                handler = new_handler or self._log_handler
+                handler = new_handler or self.log_handler
 
             oldlogformat = (self._config or {}).get('format', PatroniLogger.DEFAULT_FORMAT)
             logformat = config.get('format', PatroniLogger.DEFAULT_FORMAT)
@@ -109,17 +122,17 @@ class PatroniLogger(Thread):
                 handler.setFormatter(logging.Formatter(logformat, dateformat))
 
             if new_handler:
-                with self._log_handler_lock:
-                    if self._log_handler:
-                        self._old_handlers.append(self._log_handler)
-                    self._log_handler = new_handler
+                with self.log_handler_lock:
+                    if self.log_handler:
+                        self._old_handlers.append(self.log_handler)
+                    self.log_handler = new_handler
 
             self._config = config.copy()
             self.update_loggers()
 
     def _close_old_handlers(self):
         while True:
-            with self._log_handler_lock:
+            with self.log_handler_lock:
                 if not self._old_handlers:
                     break
                 handler = self._old_handlers.pop()
@@ -129,6 +142,11 @@ class PatroniLogger(Thread):
                 _LOGGER.exception('Failed to close the old log handler %s', handler)
 
     def run(self):
+        # switch to QueueHandler only when the thread was started
+        with self.log_handler_lock:
+            self._root_logger.addHandler(self._queue_handler)
+            self._root_logger.removeHandler(self._proxy_handler)
+
         while True:
             self._close_old_handlers()
 
@@ -136,7 +154,7 @@ class PatroniLogger(Thread):
             if record is None:
                 break
 
-            self._log_handler.handle(record)
+            self.log_handler.handle(record)
             self._queue_handler.queue.task_done()
 
     def shutdown(self):

tests/test_log.py

@@ -38,12 +38,13 @@ class TestPatroniLogger(unittest.TestCase):
         logger = PatroniLogger()
         patroni_config = Config()
         logger.reload_config(patroni_config['log'])
+        logger.start()
 
         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'])
+        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'])

tests/test_patroni.py

@@ -122,6 +122,7 @@ class TestPatroni(unittest.TestCase):
         self.p.sighup_handler()
         self.p.ha.dcs.watch = Mock(side_effect=SleepException)
         self.p.api.start = Mock()
+        self.p.logger.start = Mock()
         self.p.config._dynamic_configuration = {}
         self.assertRaises(SleepException, self.p.run)
         with patch('patroni.config.Config.set_dynamic_configuration', Mock(return_value=True)):