diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index ce340228..78c42089 100644 --- a/docs/ENVIRONMENT.rst +++ b/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 `_) - **PATRONI\_LOG\_FORMAT**: sets the log formatting string. Default value is **%(asctime)s %(levelname)s: %(message)s** (see `the LogRecord attributes `_) - **PATRONI\_LOG\_DATEFORMAT**: sets the datetime formatting string. (see the `formatTime() documentation `_) +- **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. diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index de1844a3..c9fe9b1f 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -15,6 +15,7 @@ Log - **level**: sets the general logging level. Default value is **INFO** (see `the docs for Python logging `_) - **format**: sets the log formatting string. Default value is **%(asctime)s %(levelname)s: %(message)s** (see `the LogRecord attributes `_) - **dateformat**: sets the datetime formatting string. (see the `formatTime() documentation `_) +- **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. diff --git a/patroni/__init__.py b/patroni/__init__.py index fa8db42a..8af4d6af 100644 --- a/patroni/__init__.py +++ b/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): diff --git a/patroni/api.py b/patroni/api.py index 51833108..0628303a 100644 --- a/patroni/api.py +++ b/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): diff --git a/patroni/config.py b/patroni/config.py index c982637e..4186f047 100644 --- a/patroni/config.py +++ b/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('{'): diff --git a/patroni/log.py b/patroni/log.py index 208735ae..444f2429 100644 --- a/patroni/log.py +++ b/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 diff --git a/tests/test_api.py b/tests/test_api.py index 828e3df8..17758a4c 100644 --- a/tests/test_api.py +++ b/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) diff --git a/tests/test_log.py b/tests/test_log.py index 8cee50bf..1d8e45a0 100644 --- a/tests/test_log.py +++ b/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) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index b6fdf1e2..8af5b416 100644 --- a/tests/test_patroni.py +++ b/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']