From facee0186d0ccecbf5bf6d37688d3f9da3fb706a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 7 Oct 2019 11:00:38 +0200 Subject: [PATCH] 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 | 1 + patroni/log.py | 46 ++++++++++++++++++++++++++++++------------- tests/test_log.py | 5 +++-- tests/test_patroni.py | 1 + 4 files changed, 37 insertions(+), 16 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 8af4d6af..fff80243 100644 --- a/patroni/__init__.py +++ b/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: diff --git a/patroni/log.py b/patroni/log.py index d68a7f31..c593033e 100644 --- a/patroni/log.py +++ b/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): diff --git a/tests/test_log.py b/tests/test_log.py index 1d8e45a0..76fb5704 100644 --- a/tests/test_log.py +++ b/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']) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 567331e0..36b0c42b 100644 --- a/tests/test_patroni.py +++ b/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)):