From b576e693622c3cf745fd9c444c50550d460546bc Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 23 Jun 2017 12:38:25 +0200 Subject: [PATCH 1/8] Manage pg_hba.conf via patroni config or dynamic_configuration (#458) So far Patroni was populating pg_hba.conf only when running bootstrap code and after that it was not very handy to manage it's content, because it was necessary to login to every node, change pg_hba.conf manually and run pg_ctl reload. This commit intends to fix it and give Patroni control over pg_hba.conf. It is possible to define pg_hba.conf content via `postgresql.pg_hba` in the patroni configuration file or in the `DCS/config` (dynamic configuration). If the `hba_file` is defined in the `postgresql.parameters`, Patroni will ignore `postgresql.pg_hba`. --- docs/SETTINGS.rst | 5 ++++- patroni/postgresql.py | 43 ++++++++++++++++++++++++++++++++-------- tests/test_postgresql.py | 24 ++++++++++++++-------- 3 files changed, 55 insertions(+), 17 deletions(-) diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 7a035b86..02c66a46 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -87,8 +87,11 @@ PostgreSQL - **use\_unix\_socket**: specifies that Patroni should prefer to use unix sockets to connect to the cluster. Default value is ``false``. If ``unix_socket_directories`` is definded, Patroni will use first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that default value should be used and omit ``host`` from connection parameters. - **pgpass**: path to the `.pgpass `__ password file. Patroni creates this file before executing pg\_basebackup, the post_init script and under some other circumstances. The location must be writable by Patroni. - **recovery\_conf**: additional configuration settings written to recovery.conf when configuring follower. -- **custom_conf** : path to an optional custom ``postgresql.conf`` file, that will be used in place of ``postgresql.base.conf``. The file must exist on all cluster nodes, be readable by PostgreSQL and will be included from its location on the real ``postgresql.conf``. Note that Patroni will not monitor this file for changes, nor backup it. However, its settings can still be overriden by Patroni's own configuration facilities - see `dynamic configuration `__ for details. +- **custom\_conf** : path to an optional custom ``postgresql.conf`` file, that will be used in place of ``postgresql.base.conf``. The file must exist on all cluster nodes, be readable by PostgreSQL and will be included from its location on the real ``postgresql.conf``. Note that Patroni will not monitor this file for changes, nor backup it. However, its settings can still be overriden by Patroni's own configuration facilities - see `dynamic configuration `__ for details. - **parameters**: list of configuration settings for Postgres. Many of these are required for replication to work. +- **pg\_hba**: list of lines that Patroni will use to generate ``pg_hba.conf``. This parameter has higher priority than ``bootstrap.pg_hba``. Together with `dynamic configuration `__ it simplifies management of ``pg_hba.conf``. + - **- host all all 0.0.0.0/0 md5**. + - **- host replication replicator 127.0.0.1/32 md5**: A line like this is required for replication. - **pg\_ctl\_timeout**: How long should pg_ctl wait when doing ``start``, ``stop`` or ``restart``. Default value is 60 seconds. - **use\_pg\_rewind**: try to use pg\_rewind on the former leader when it joins cluster as a replica. - **remove\_data\_directory\_on\_rewind\_failure**: If this option is enabled, Patroni will remove postgres data directory and recreate replica. Otherwise it will try to follow the new leader. Default value is **false**. diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 6a8d1d4b..e8dbb90c 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -157,6 +157,8 @@ class Postgresql(object): self.set_state('running') self.set_role('master' if self.is_leader() else 'replica') self._write_postgresql_conf() # we are "joining" already running postgres + if self._replace_pg_hba(): + self.reload() @property def _configuration_to_save(self): @@ -280,7 +282,7 @@ class Postgresql(object): self._superuser = config['authentication'].get('superuser', {}) server_parameters = self.get_server_parameters(config) - local_connection_address_changed = pending_reload = pending_restart = False + conf_changed = hba_changed = local_connection_address_changed = pending_restart = False if self.state == 'running': changes = {p: v for p, v in server_parameters.items() if '.' not in p} changes.update({p: None for p, v in self._server_parameters.items() if not ('.' in p or p in changes)}) @@ -308,24 +310,27 @@ class Postgresql(object): or r[0] in ('listen_addresses', 'port'): local_connection_address_changed = True else: - pending_reload = True + conf_changed = True for param in changes: if param in server_parameters: logger.warning('Removing invalid parameter `%s` from postgresql.parameters', param) server_parameters.pop(param) # Check that user-defined-paramters have changed (parameters with period in name) - if not pending_reload: + if not conf_changed: for p, v in server_parameters.items(): if '.' in p and (p not in self._server_parameters or str(v) != str(self._server_parameters[p])): - pending_reload = True + conf_changed = True break - if not pending_reload: + if not conf_changed: for p, v in self._server_parameters.items(): if '.' in p and (p not in server_parameters or str(v) != str(server_parameters[p])): - pending_reload = True + conf_changed = True break + if not config['parameters'].get('hba_file') and config.get('pg_hba'): + hba_changed = self.config.get('pg_hba', []) != config['pg_hba'] + self.config = config self._pending_restart = pending_restart self._server_parameters = server_parameters @@ -334,9 +339,15 @@ class Postgresql(object): if not local_connection_address_changed: self.resolve_connection_addresses() - if pending_reload: + if conf_changed: self._write_postgresql_conf() + + if hba_changed: + self._replace_pg_hba() + + if conf_changed or hba_changed: self.reload() + self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout']/2.0 @property @@ -501,7 +512,8 @@ class Postgresql(object): if pwfile: os.remove(pwfile) if ret: - self.write_pg_hba(config.get('pg_hba', [])) + if not self.config['parameters'].get('hba_file') and not self.config.get('pg_hba'): + self.write_pg_hba(config.get('pg_hba', [])) self._major_version = self.get_major_version() self._server_parameters = self.get_server_parameters(self.config) else: @@ -789,6 +801,7 @@ class Postgresql(object): self._pending_restart = False self._write_postgresql_conf() + self._replace_pg_hba() self.resolve_connection_addresses() opts = {p: self._server_parameters[p] for p in self.CMDLINE_OPTIONS if p in self._server_parameters} @@ -1060,6 +1073,20 @@ class Postgresql(object): with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'a') as f: f.write('\n{}\n'.format('\n'.join(config))) + def _replace_pg_hba(self): + """ + Replace pg_hba.conf content in the PGDATA if hba_file is not defined in the + `postgresql.parameters` and pg_hba is defined in `postgresql` configuration section. + + :returns: True if pg_hba.conf was rewritten. + """ + if not self.config['parameters'].get('hba_file') and self.config.get('pg_hba'): + with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'w') as f: + f.write('# Do not edit this file manually!\n# It will be overwritten by Patroni!\n') + for line in self.config['pg_hba']: + f.write('{0}\n'.format(line)) + return True + def primary_conninfo(self, member): if not (member and member.conn_url) or member.name == self.name: return None diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index a7a477bd..b2376d7b 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -178,6 +178,7 @@ class TestPostgresql(unittest.TestCase): 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'parameters': self._PARAMETERS, 'recovery_conf': {'foo': 'bar'}, + 'pg_hba': ['host all all 0.0.0.0/0 md5'], 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_restart': 'true', 'on_role_change': 'true', 'on_reload': 'true' @@ -498,15 +499,22 @@ class TestPostgresql(unittest.TestCase): with patch.object(Postgresql, 'run_bootstrap_post_init', Mock(return_value=False)): self.assertRaises(PostgresException, self.p.bootstrap, {}) - self.p.bootstrap({'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}, - 'pg_hba': ['host replication replicator 127.0.0.1/32 md5', - 'hostssl all all 0.0.0.0/0 md5', - 'host all all 0.0.0.0/0 md5'], - 'post_init': '/bin/false'}) + config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}} + + self.p.bootstrap(config) with open(os.path.join(self.data_dir, 'pg_hba.conf')) as f: lines = f.readlines() - assert 'host replication replicator 127.0.0.1/32 md5\n' in lines - assert 'host all all 0.0.0.0/0 md5\n' in lines + self.assertTrue('host all all 0.0.0.0/0 md5\n' in lines) + + self.p.config.pop('pg_hba') + config.update({'post_init': '/bin/false', + 'pg_hba': ['host replication replicator 127.0.0.1/32 md5', + 'hostssl all all 0.0.0.0/0 md5', + 'host all all 0.0.0.0/0 md5']}) + self.p.bootstrap(config) + with open(os.path.join(self.data_dir, 'pg_hba.conf')) as f: + lines = f.readlines() + self.assertTrue('host replication replicator 127.0.0.1/32 md5\n' in lines) def test_run_bootstrap_post_init(self): with patch('subprocess.call', Mock(return_value=1)): @@ -593,7 +601,7 @@ class TestPostgresql(unittest.TestCase): def test_reload_config(self): parameters = self._PARAMETERS.copy() parameters.pop('f.oo') - config = {'use_unix_socket': True, 'authentication': {}, + config = {'pg_hba': [''], 'use_unix_socket': True, 'authentication': {}, 'retry_timeout': 10, 'listen': '*', 'parameters': parameters} self.p.reload_config(config) parameters['b.ar'] = 'bar' From fa68eba33ea18cbb0754f48784b589fdfe9b0c0b Mon Sep 17 00:00:00 2001 From: jouir Date: Mon, 3 Jul 2017 12:07:09 +0200 Subject: [PATCH 2/8] Bugfix sleep func missing in Kazoo client (#464) This patch is adding the sleep function needed in the connection_retry and command_retry parameters. The KazooClient is trying to compare them with the ones included in the handler at instantiation. --- patroni/dcs/zookeeper.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index c68bd3d8..1215d820 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -1,6 +1,7 @@ import logging +import time -from kazoo.client import KazooClient, KazooState +from kazoo.client import KazooClient, KazooState, KazooRetry from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.handlers.threading import SequentialThreadingHandler from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState @@ -51,8 +52,9 @@ class ZooKeeper(AbstractDCS): hosts = ','.join(hosts) self._client = KazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']), - timeout=config['ttl'], connection_retry={'max_delay': 1, 'max_tries': -1}, - command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1}) + timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1, + sleep_func=time.sleep), command_retry=KazooRetry(deadline=config['retry_timeout'], + max_delay=1, max_tries=-1, sleep_func=time.sleep)) self._client.add_listener(self.session_listener) self._my_member_data = None From 10f2321334e8ac8f9774ed6b6cbb7867b182b201 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 3 Jul 2017 12:07:19 +0200 Subject: [PATCH 3/8] Don't fail if one of DCS implementation can't be loaded (#463) It might be that it's not required by configuration. --- patroni/dcs/__init__.py | 31 +++++++++++++++++++------------ tests/test_ha.py | 1 + 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index ce4020af..df8901a2 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -3,6 +3,7 @@ import dateutil import importlib import inspect import json +import logging import os import pkgutil import six @@ -14,6 +15,8 @@ from random import randint from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl from threading import Event, Lock +logger = logging.getLogger(__name__) + def parse_connection_string(value): """Original Governor stores connection strings for each cluster members if a following format: @@ -51,18 +54,22 @@ def dcs_modules(): def get_dcs(config): available_implementations = set() for module_name in dcs_modules(): - module = importlib.import_module(module_name) - for name in filter(lambda name: not name.startswith('__'), dir(module)): # iterate through module content - value = getattr(module, name) - name = name.lower() - # try to find implementation of AbstractDCS interface, class name must match with module_name - if inspect.isclass(value) and issubclass(value, AbstractDCS) and __package__ + '.' + name == module_name: - available_implementations.add(name) - if name in config: # which has configuration section in the config file - # propagate some parameters - config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', - 'patronictl', 'ttl', 'retry_timeout') if p in config}) - return value(config[name]) + try: + module = importlib.import_module(module_name) + for name in filter(lambda name: not name.startswith('__'), dir(module)): # iterate through module content + item = getattr(module, name) + name = name.lower() + # try to find implementation of AbstractDCS interface, class name must match with module_name + if inspect.isclass(item) and issubclass(item, AbstractDCS) and __package__ + '.' + name == module_name: + available_implementations.add(name) + if name in config: # which has configuration section in the config file + # propagate some parameters + config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', + 'patronictl', 'ttl', 'retry_timeout') if p in config}) + return item(config[name]) + except ImportError: + if not config.get('patronictl'): + logger.info('Failed to import %s', module_name) raise PatroniException("""Can not find suitable configuration of distributed configuration store Available implementations: """ + ', '.join(available_implementations)) diff --git a/tests/test_ha.py b/tests/test_ha.py index 71d33d25..0a899823 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -135,6 +135,7 @@ class TestHa(unittest.TestCase): @patch('socket.getaddrinfo', socket_getaddrinfo) @patch('psycopg2.connect', psycopg2_connect) + @patch('patroni.dcs.dcs_modules', Mock(return_value=['foo', 'patroni.dcs.etcd'])) @patch.object(etcd.Client, 'read', etcd_read) def setUp(self): with patch.object(Client, 'machines') as mock_machines: From 23e6f65156256b45fe55c1508ae7e20303831a03 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 3 Jul 2017 12:07:24 +0200 Subject: [PATCH 4/8] Don't wal_keep_segments as command line argument to postgres (#460) It make it not possible to change it without restart. Fixes https://github.com/zalando/patroni/issues/459 --- patroni/postgresql.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index e8dbb90c..aa2956f9 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -804,8 +804,8 @@ class Postgresql(object): self._replace_pg_hba() self.resolve_connection_addresses() - opts = {p: self._server_parameters[p] for p in self.CMDLINE_OPTIONS if p in self._server_parameters} - options = ['--{0}={1}'.format(p, v) for p, v in opts.items()] + options = ['--{0}={1}'.format(p, self._server_parameters[p]) for p in self.CMDLINE_OPTIONS + if p in self._server_parameters and p != 'wal_keep_segments'] start_initiated = time.time() From b60f65a2caceaeed9bba6d87b0056ce7d7f1adfa Mon Sep 17 00:00:00 2001 From: jouir Date: Tue, 4 Jul 2017 16:13:18 +0200 Subject: [PATCH 5/8] Compatibility with old kazoo (#468) ```python AttributeError: 'KazooClient' object has no attribute '_retry' ``` Fixes #467 --- patroni/dcs/zookeeper.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 1215d820..82cdff4b 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -116,7 +116,8 @@ class ZooKeeper(AbstractDCS): return True def set_retry_timeout(self, retry_timeout): - self._client._retry.deadline = retry_timeout + retry = self._client.retry if isinstance(self._client.retry, KazooRetry) else self._client._retry + retry.deadline = retry_timeout def get_node(self, key, watch=None): try: From 4ca94a5dabd7500b6f67a9ad5b2cd7776baf2c7b Mon Sep 17 00:00:00 2001 From: jouir Date: Tue, 4 Jul 2017 16:14:17 +0200 Subject: [PATCH 6/8] Add config_dir option for configuration files location (#466) On debian, the configuration files (postgresql.conf, pg_hba.conf, etc) are not stored in the data directory. It would be great to be able to configure the location of this separate directory. Patroni could override existing configuration files where they are used to be. The default is to store configuration files in the data directory. This setting is targeting custom installations like debian and any others moving configuration files out of the data directory. Fixes #465 --- docs/ENVIRONMENT.rst | 1 + docs/SETTINGS.rst | 1 + patroni/postgresql.py | 35 ++++++++++++++++++++++------------- postgres0.yml | 1 + postgres1.yml | 1 + postgres2.yml | 1 + tests/test_config.py | 1 + tests/test_postgresql.py | 6 ++++-- 8 files changed, 32 insertions(+), 15 deletions(-) diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index c446b725..3a8494f6 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -46,6 +46,7 @@ PostgreSQL - **PATRONI\_POSTGRESQL\_LISTEN**: IP address + port that Postgres listens to. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node. - **PATRONI\_POSTGRESQL\_CONNECT\_ADDRESS**: IP address + port through which Postgres is accessible from other nodes and applications. - **PATRONI\_POSTGRESQL\_DATA\_DIR**: The location of the Postgres data directory, either existing or to be initialized by Patroni. +- **PATRONI\_POSTGRESQL\_CONFIG\_DIR**: The location of the Postgres configuration directory, defaults to the data directory. Must be writable by Patroni. - **PATRONI\_POSTGRESQL\_BIN_DIR**: Path to PostgreSQL binaries. (pg_ctl, pg_rewind, pg_basebackup, postgres) The default value is an empty string meaning that PATH environment variable will be used to find the executables. - **PATRONI\_POSTGRESQL\_PGPASS**: path to the `.pgpass `__ password file. Patroni creates this file before executing pg\_basebackup and under some other circumstances. The location must be writable by Patroni. - **PATRONI\_REPLICATION\_USERNAME**: replication username; the user will be created during initialization. Replicas will use this user to access master via streaming replication diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 02c66a46..fac8a90f 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -82,6 +82,7 @@ PostgreSQL - **connect\_address**: IP address + port through which Postgres is accessible from other nodes and applications. - **create\_replica\_methods**: an ordered list of the create methods for turning a Patroni node into a new replica. "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured as its own config item. - **data\_dir**: The location of the Postgres data directory, either existing or to be initialized by Patroni. +- **config\_dir**: The location of the Postgres configuration directory, defaults to the data directory. Must be writable by Patroni. - **bin\_dir**: Path to PostgreSQL binaries. (pg_ctl, pg_rewind, pg_basebackup, postgres) The default value is an empty string meaning that PATH environment variable will be used to find the executables. - **listen**: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node. - **use\_unix\_socket**: specifies that Patroni should prefer to use unix sockets to connect to the cluster. Default value is ``false``. If ``unix_socket_directories`` is definded, Patroni will use first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that default value should be used and omit ``host`` from connection parameters. diff --git a/patroni/postgresql.py b/patroni/postgresql.py index aa2956f9..fb3f9b2b 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -99,6 +99,7 @@ class Postgresql(object): self._bin_dir = config.get('bin_dir') or '' self._database = config.get('database', 'postgres') self._data_dir = config['data_dir'] + self._config_dir = config.get('config_dir') or self._data_dir self._pending_restart = False self.__thread_ident = current_thread().ident @@ -120,9 +121,9 @@ class Postgresql(object): self.__cb_called = False self.__cb_pending = None config_base_name = config.get('config_base_name', 'postgresql') - self._postgresql_conf = os.path.join(self._data_dir, config_base_name + '.conf') + self._postgresql_conf = os.path.join(self._config_dir, config_base_name + '.conf') self._postgresql_base_conf_name = config_base_name + '.base.conf' - self._postgresql_base_conf = os.path.join(self._data_dir, self._postgresql_base_conf_name) + self._postgresql_base_conf = os.path.join(self._config_dir, self._postgresql_base_conf_name) self._recovery_conf = os.path.join(self._data_dir, 'recovery.conf') self._postmaster_pid = os.path.join(self._data_dir, 'postmaster.pid') self._trigger_file = config.get('recovery_conf', {}).get('trigger_file') or 'promote' @@ -162,11 +163,11 @@ class Postgresql(object): @property def _configuration_to_save(self): - configuration = [self._postgresql_conf] + configuration = [os.path.basename(self._postgresql_conf)] if 'custom_conf' not in self.config: - configuration.append(self._postgresql_base_conf) + configuration.append(os.path.basename(self._postgresql_base_conf)) if not self.config['parameters'].get('hba_file'): - configuration.append(os.path.join(self._data_dir, 'pg_hba.conf')) + configuration.append('pg_hba.conf') return configuration @property @@ -826,8 +827,9 @@ class Postgresql(object): return False start_initiated = time.time() - proc = call_self(['pg_ctl_start', self._pgcommand('postgres'), '-D', self._data_dir] + options, - close_fds=True, preexec_fn=os.setsid, stdout=subprocess.PIPE, + proc = call_self(['pg_ctl_start', self._pgcommand('postgres'), '-D', self._data_dir, + '--config-file={}'.format(self._postgresql_conf)] + options, close_fds=True, + preexec_fn=os.setsid, stdout=subprocess.PIPE, env={p: os.environ[p] for p in ('PATH', 'LC_ALL', 'LANG') if p in os.environ}) pid = int(proc.stdout.readline().strip()) proc.wait() @@ -1060,6 +1062,9 @@ class Postgresql(object): with open(self._postgresql_conf, 'w') as f: f.write('# Do not edit this file manually!\n# It will be overwritten by Patroni!\n') f.write("include '{0}'\n\n".format(self.config.get('custom_conf') or self._postgresql_base_conf_name)) + f.write("data_directory = '{}'\n".format(self._data_dir)) + f.write("hba_file = '{}'\n".format(os.path.join(self._config_dir, 'pg_hba.conf'))) + f.write("ident_file = '{}'\n".format(os.path.join(self._config_dir, 'pg_ident.conf'))) for name, value in sorted(self._server_parameters.items()): f.write("{0} = '{1}'\n".format(name, value)) @@ -1070,7 +1075,7 @@ class Postgresql(object): return True def write_pg_hba(self, config): - with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'a') as f: + with open(os.path.join(self._config_dir, 'pg_hba.conf'), 'a') as f: f.write('\n{}\n'.format('\n'.join(config))) def _replace_pg_hba(self): @@ -1081,7 +1086,7 @@ class Postgresql(object): :returns: True if pg_hba.conf was rewritten. """ if not self.config['parameters'].get('hba_file') and self.config.get('pg_hba'): - with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'w') as f: + with open(os.path.join(self._config_dir, 'pg_hba.conf'), 'w') as f: f.write('# Do not edit this file manually!\n# It will be overwritten by Patroni!\n') for line in self.config['pg_hba']: f.write('{0}\n'.format(line)) @@ -1366,8 +1371,10 @@ class Postgresql(object): """ try: for f in self._configuration_to_save: - if os.path.isfile(f): - shutil.copy(f, f + '.backup') + config_file = os.path.join(self._config_dir, f) + backup_file = os.path.join(self._data_dir, f + '.backup') + if os.path.isfile(config_file): + shutil.copy(config_file, backup_file) except IOError: logger.exception('unable to create backup copies of configuration files') @@ -1375,8 +1382,10 @@ class Postgresql(object): """ restore a previously saved postgresql.conf """ try: for f in self._configuration_to_save: - if not os.path.isfile(f) and os.path.isfile(f + '.backup'): - shutil.copy(f + '.backup', f) + config_file = os.path.join(self._config_dir, f) + backup_file = os.path.join(self._data_dir, f + '.backup') + if not os.path.isfile(config_file) and os.path.isfile(backup_file): + shutil.copy(backup_file, config_file) except IOError: logger.exception('unable to restore configuration files from backup') diff --git a/postgres0.yml b/postgres0.yml index f44110da..1cb634d3 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -66,6 +66,7 @@ postgresql: connect_address: 127.0.0.1:5432 data_dir: data/postgresql0 # bin_dir: +# config_dir: pgpass: /tmp/pgpass0 authentication: replication: diff --git a/postgres1.yml b/postgres1.yml index 5405ab67..9c1d141b 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -64,6 +64,7 @@ postgresql: connect_address: 127.0.0.1:5433 data_dir: data/postgresql1 # bin_dir: +# config_dir: pgpass: /tmp/pgpass1 authentication: replication: diff --git a/postgres2.yml b/postgres2.yml index ec4a7c3a..b77f5bec 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -61,6 +61,7 @@ postgresql: connect_address: 127.0.0.1:5434 data_dir: data/postgresql2 # bin_dir: +# config_dir: pgpass: /tmp/pgpass2 authentication: replication: diff --git a/tests/test_config.py b/tests/test_config.py index 21d19dae..9e7ddafa 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -39,6 +39,7 @@ class TestConfig(unittest.TestCase): 'PATRONI_POSTGRESQL_LISTEN': '0.0.0.0:5432', 'PATRONI_POSTGRESQL_CONNECT_ADDRESS': '127.0.0.1:5432', 'PATRONI_POSTGRESQL_DATA_DIR': 'data/postgres0', + 'PATRONI_POSTGRESQL_CONFIG_DIR': 'data/postgres0', 'PATRONI_POSTGRESQL_PGPASS': '/tmp/pgpass0', 'PATRONI_ETCD_HOST': '127.0.0.1:2379', 'PATRONI_ETCD_URL': 'https://127.0.0.1:2379', diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index b2376d7b..81988bf4 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -168,9 +168,11 @@ class TestPostgresql(unittest.TestCase): @patch.object(Postgresql, 'is_running', Mock(return_value=True)) def setUp(self): self.data_dir = 'data/test0' + self.config_dir = self.data_dir if not os.path.exists(self.data_dir): os.makedirs(self.data_dir) - self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir, 'retry_timeout': 10, + self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir, + 'config_dir': self.config_dir, 'retry_timeout': 10, 'listen': '127.0.0.2, 127.0.0.3:5432', 'connect_address': '127.0.0.2:5432', 'authentication': {'superuser': {'username': 'test', 'password': 'test'}, 'replication': {'username': 'replicator', 'password': 'rep-pass'}}, @@ -502,7 +504,7 @@ class TestPostgresql(unittest.TestCase): config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}} self.p.bootstrap(config) - with open(os.path.join(self.data_dir, 'pg_hba.conf')) as f: + with open(os.path.join(self.config_dir, 'pg_hba.conf')) as f: lines = f.readlines() self.assertTrue('host all all 0.0.0.0/0 md5\n' in lines) From 0b2134aba35d4118167f277c137cb5be4790d1d6 Mon Sep 17 00:00:00 2001 From: Nick Stott Date: Wed, 5 Jul 2017 03:56:14 -0400 Subject: [PATCH 7/8] truncate the fqdn to 64 chars, or NAMEDATALEN-1 (#470) as described here, https://github.com/postgres/postgres/blob/master/src/backend/replication/slot.c#L164-L210 the slot name should be truncated to 63 chars, getting an error related to a slot_name ``` FATAL: replication slot name "c_formationid4_main_m_1_c_formationid4_main_m_nicksaccount_svc_c" is too long ``` the leader has the following slot names ``` postgres=# select * from pg_replication_slots; slot_name | plugin | slot_type | datoid | database | active | active_pid | xmin | catalog_xmin | restart_ls n | confirmed_flush_lsn -----------------------------------------------------------------+--------+-----------+--------+----------+--------+------------+------+--------------+----------- --+--------------------- c_formationid4_main_m_1_c_formationid4_main_m_nicksaccount_svc_ | | physical | | | f | | | | | c_formationid4_main_m_2_c_formationid4_main_m_nicksaccount_svc_ | | physical | | | f | | | | | (2 rows) ``` --- patroni/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index fb3f9b2b..9d31f9d1 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -56,7 +56,7 @@ def slot_name_from_member_name(member_name): return '_' if c in '-.' else "u{:04d}".format(ord(c)) slot_name = re.sub('[^a-z0-9_]', replace_char, member_name.lower()) - return slot_name[0:64] + return slot_name[0:63] class Postgresql(object): From acc6d7c2c210111661401000bc0c86356af9c65d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 11 Jul 2017 10:00:30 +0200 Subject: [PATCH 8/8] Watchdog unit-tests, bugfixes and questions (#449) Implement missing unit-tests for and drop unused code --- features/environment.py | 8 ++- features/steps/patroni_api.py | 3 +- features/steps/watchdog.py | 2 +- patroni/postgresql.py | 54 +----------------- patroni/watchdog/base.py | 4 +- patroni/watchdog/linux.py | 4 +- tests/test_async_executor.py | 10 +++- tests/test_ha.py | 32 +++++++++-- tests/test_postgresql.py | 64 +++++++++++++++++++-- tests/test_watchdog.py | 103 +++++++++++++++++++++++++++++----- 10 files changed, 198 insertions(+), 86 deletions(-) diff --git a/features/environment.py b/features/environment.py index 81dfb9f2..8377ee79 100644 --- a/features/environment.py +++ b/features/environment.py @@ -81,6 +81,7 @@ class AbstractController(object): def cancel_background(self): pass + class PatroniController(AbstractController): __PORT = 5440 PATRONI_CONFIG = '{}.yml' @@ -275,6 +276,7 @@ class PatroniController(AbstractController): if 'process' not in p.cmdline()[0]: p.terminate() + class ProcessHang(object): """A background thread implementing a cancelable process hang via SIGSTOP.""" @@ -499,7 +501,8 @@ class PatroniPoolController(object): def start(self, name, max_wait_limit=20, tags=None, with_watchdog=False): if name not in self._processes: - self._processes[name] = PatroniController(self._context, name, self.patroni_path, self._output_dir, tags, with_watchdog=with_watchdog) + self._processes[name] = PatroniController(self._context, name, self.patroni_path, + self._output_dir, tags, with_watchdog) self._processes[name].start(max_wait_limit) def __getattr__(self, func): @@ -540,7 +543,7 @@ class WatchdogMonitor(object): def __init__(self, name, work_directory, output_dir): self.fifo_path = os.path.join(work_directory, 'data', 'watchdog.{0}.fifo'.format(name)) self.fifo_file = None - self._stop_requested = False # Relying on bool setting being atomic + self._stop_requested = False # Relying on bool setting being atomic self._thread = None self.last_ping = None self.was_pinged = False @@ -637,7 +640,6 @@ class WatchdogMonitor(object): self._thread.join() self._thread = None - def reset(self): self._log("reset") self.was_pinged = self.was_closed = self._was_triggered = False diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index ca554754..e6734f36 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -111,7 +111,8 @@ def check_response(context, component, data): assert context.status_code == int(data),\ "status code {0} != {1}, response: {2}".format(context.status_code, data, context.response) elif component == 'returncode': - assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code, data, context.response) + assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code, + data, context.response) elif component == 'text': assert context.response == data.strip('"'), "response {0} does not contain {1}".format(context.response, data) elif component == 'output': diff --git a/features/steps/watchdog.py b/features/steps/watchdog.py index 255a7391..e37ace3d 100644 --- a/features/steps/watchdog.py +++ b/features/steps/watchdog.py @@ -1,6 +1,7 @@ from behave import step, then import time + def polling_loop(timeout, interval=1): """Returns an iterator that returns values until timeout has passed. Timeout is measured from start of iteration.""" start_time = time.time() @@ -12,7 +13,6 @@ def polling_loop(timeout, interval=1): time.sleep(interval) - @step('I start {name:w} with watchdog') def start_patroni_with_watchdog(context, name): return context.pctl.start(name, with_watchdog=True) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 9d31f9d1..6fce3018 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -954,13 +954,9 @@ class Postgresql(object): def _wait_for_connection_close(self, pid): try: with self.connection().cursor() as cur: - while True: # Need a timeout here? - if pid == self.get_pid() and self.is_pid_running(pid): - cur.execute("SELECT 1") - time.sleep(STOP_POLLING_INTERVAL) - continue - else: - break + while pid == self.get_pid() and self.is_pid_running(pid): # Need a timeout here? + cur.execute("SELECT 1") + time.sleep(STOP_POLLING_INTERVAL) except psycopg2.Error: pass @@ -1319,50 +1315,6 @@ class Postgresql(object): self.call_nowait(ACTION_ON_ROLE_CHANGE) return True - def _do_rewind(self, leader): - logger.info("rewind flag is set") - - if self.is_running() and not self.stop(checkpoint=False): - logger.warning('Can not run pg_rewind because postgres is still running') - return False - - # prepare pg_rewind connection - r = leader.conn_kwargs(self._superuser) - - # first make sure that we are really trying to rewind - # from the master and run a checkpoint on a t in order to - # make it store the new timeline (5540277D.8020309@iki.fi) - leader_status = self.checkpoint(r) - if leader_status: - logger.warning('Can not use %s for rewind: %s', leader.name, leader_status) - return False - - # at present, pg_rewind only runs when the cluster is shut down cleanly - # and not shutdown in recovery. We have to remove the recovery.conf if present - # and start/shutdown in a single user mode to emulate this. - # XXX: if recovery.conf is linked, it will be written anew as a normal file. - if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf): - os.unlink(self._recovery_conf) - - # Archived segments might be useful to pg_rewind, - # clean the flags that tell we should remove them. - self.cleanup_archive_status() - - # Start in a single user mode and stop to produce a clean shutdown - opts = self.read_postmaster_opts() - opts.update({'archive_mode': 'on', 'archive_command': 'false'}) - self.single_user_mode(options=opts) - - try: - if not self.rewind(r): - logger.error('unable to rewind the former master') - if self.config.get('remove_data_directory_on_rewind_failure', False): - self.remove_data_directory() - return False - return True - finally: - self._need_rewind = False - def save_configuration_files(self): """ copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files diff --git a/patroni/watchdog/base.py b/patroni/watchdog/base.py index 6154d583..da65155d 100644 --- a/patroni/watchdog/base.py +++ b/patroni/watchdog/base.py @@ -94,7 +94,7 @@ class Watchdog(object): logger.info("{0} activated with {1} second timeout, timing slack {2} seconds" .format(self.impl.describe(), actual_timeout, slack)) else: - if self.mode == MODE_REQUIRED: + if self.mode == MODE_REQUIRED: # XXX: can we really get here? logger.error("Configuration requires watchdog, but watchdog could not be activated") sys.exit(1) @@ -116,7 +116,7 @@ class Watchdog(object): logger.error("Error while sending keepalive: %s", e) def _get_impl(self): - if self.mode not in [MODE_AUTOMATIC, MODE_REQUIRED]: + if self.mode not in [MODE_AUTOMATIC, MODE_REQUIRED]: # XXX: can't be reached return NullWatchdog() if self.driver == 'testing': diff --git a/patroni/watchdog/linux.py b/patroni/watchdog/linux.py index e9cc3155..92f0db0e 100644 --- a/patroni/watchdog/linux.py +++ b/patroni/watchdog/linux.py @@ -145,7 +145,7 @@ class LinuxWatchdogDevice(WatchdogBase): os.close(self._fd) self._fd = None except OSError as e: - return WatchdogError("Error while closing {0}: {1}".format(self.describe(), e)) + raise WatchdogError("Error while closing {0}: {1}".format(self.describe(), e)) @property def can_be_disabled(self): @@ -176,7 +176,7 @@ class LinuxWatchdogDevice(WatchdogBase): try: _, version, identity = self.get_support() ver_str = " (firmware {0})".format(version) if version else "" - except WatchdogError: + except WatchdogError: # XXX: Can it really be raise when self._fd is not None? pass return identity + ver_str + dev_str diff --git a/tests/test_async_executor.py b/tests/test_async_executor.py index 6f867428..2c726c0e 100644 --- a/tests/test_async_executor.py +++ b/tests/test_async_executor.py @@ -1,7 +1,7 @@ import unittest from mock import Mock, patch -from patroni.async_executor import AsyncExecutor +from patroni.async_executor import AsyncExecutor, CriticalTask from threading import Thread @@ -16,3 +16,11 @@ class TestAsyncExecutor(unittest.TestCase): def test_run(self): self.a.run(Mock(side_effect=Exception())) + + +class TestCriticalTask(unittest.TestCase): + + def test_completed_task(self): + ct = CriticalTask() + ct.complete(1) + self.assertFalse(ct.cancel()) diff --git a/tests/test_ha.py b/tests/test_ha.py index 0a899823..bce463f2 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1,6 +1,7 @@ import datetime import etcd import os +import time import unittest from mock import Mock, MagicMock, PropertyMock, patch @@ -8,12 +9,13 @@ from patroni.config import Config from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState from patroni.dcs.etcd import Client from patroni.exceptions import DCSError, PostgresException -from patroni.ha import Ha, _MemberStatus +from patroni.ha import Ha, _MemberStatus, BackgroundKeepaliveSender from patroni.postgresql import Postgresql from patroni.watchdog import Watchdog from patroni.utils import tzutc from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get from test_postgresql import psycopg2_connect +from threading import Event def true(*args, **kwargs): @@ -246,7 +248,8 @@ class TestHa(unittest.TestCase): def test_demote_because_not_having_lock(self): self.ha.cluster.is_unlocked = false - self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader') + with patch.object(Watchdog, 'is_running', PropertyMock(return_value=True)): + self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader') def test_demote_because_update_lock_failed(self): self.ha.cluster.is_unlocked = false @@ -335,6 +338,7 @@ class TestHa(unittest.TestCase): with patch.object(self.ha, "restart_matches", return_value=False): self.assertEquals(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied")) + @patch('os.kill', Mock()) def test_restart_in_progress(self): with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)): self.ha.restart({}, run_async=True) @@ -349,9 +353,10 @@ class TestHa(unittest.TestCase): self.ha.update_lock = false self.p.set_role('master') - with patch('patroni.postgresql.Postgresql.stop') as stop_mock: - self.assertEquals(self.ha.run_cycle(), 'lost leader lock during restart') - stop_mock.assert_called() + with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)): + with patch('patroni.postgresql.Postgresql.stop') as stop_mock: + self.assertEquals(self.ha.run_cycle(), 'lost leader lock during restart') + stop_mock.assert_called() @patch('requests.get', requests_get) def test_manual_failover_from_leader(self): @@ -797,6 +802,10 @@ class TestHa(unittest.TestCase): def test_wakup(self): self.ha.wakeup() + def test_shutdown(self): + self.p.is_running = false + self.ha.shutdown() + @patch('time.sleep', Mock()) def test_leader_with_empty_directory(self): self.ha.cluster = get_cluster_initialized_with_leader() @@ -808,3 +817,16 @@ class TestHa(unittest.TestCase): self.ha.has_lock = false # will not say bootstrap from leader as replica can't self elect self.assertEquals(self.ha.run_cycle(), "trying to bootstrap from replica 'other'") + + +class TestBackgroundKeepaliveSender(unittest.TestCase): + + def test_run(self): + safe_event = Event() + ha = Mock() + ha.dcs.loop_wait = 0.1 + with BackgroundKeepaliveSender(ha, safe_event): + time.sleep(1) + safe_event.set() + time.sleep(1) + self.assertTrue(ha.keepalive.call_count > 2) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 81988bf4..e8e024fa 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,3 +1,4 @@ +import errno import mock # for the mock.call method, importing it without a namespace breaks python3 import os import psycopg2 @@ -6,6 +7,7 @@ import subprocess import unittest from mock import Mock, MagicMock, PropertyMock, patch, mock_open +from patroni.async_executor import CriticalTask from patroni.dcs import Cluster, Leader, Member, SyncState from patroni.exceptions import PostgresException, PostgresConnectionException from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE @@ -216,13 +218,13 @@ class TestPostgresql(unittest.TestCase): mock_is_running.return_value = True mock_wait_for_port_open.return_value = True mock_wait_for_startup.return_value = False - mock_popen.stdout.readline.return_value = '123' + mock_popen.return_value.stdout.readline.return_value = '123' self.assertTrue(self.p.start()) mock_is_running.return_value = False open(os.path.join(self.data_dir, 'postmaster.pid'), 'w').close() pg_conf = os.path.join(self.data_dir, 'postgresql.conf') open(pg_conf, 'w').close() - self.assertFalse(self.p.start()) + self.assertFalse(self.p.start(task=CriticalTask())) with open(pg_conf) as f: lines = f.readlines() self.assertTrue("f.oo = 'bar'\n" in lines) @@ -233,6 +235,9 @@ class TestPostgresql(unittest.TestCase): mock_wait_for_port_open.return_value = False self.assertFalse(self.p.start()) + task = CriticalTask() + task.cancel() + self.assertFalse(self.p.start(task=task)) @patch.object(Postgresql, 'pg_isready') @patch.object(Postgresql, 'read_pid_file') @@ -266,13 +271,24 @@ class TestPostgresql(unittest.TestCase): mock_pg_isready.return_value = 'garbage' self.assertTrue(self.p.wait_for_port_open(42, 100., 1)) + @patch('time.sleep', Mock()) @patch.object(Postgresql, 'is_running') - def test_stop(self, mock_is_running): + @patch.object(Postgresql, 'get_pid') + def test_stop(self, mock_get_pid, mock_is_running): mock_is_running.return_value = True + mock_get_pid.return_value = 0 self.assertTrue(self.p.stop()) - with patch('subprocess.call', Mock(return_value=1)): - mock_is_running.return_value = False + mock_get_pid.return_value = -1 + self.assertFalse(self.p.stop()) + mock_get_pid.return_value = 123 + with patch('os.kill', Mock(side_effect=[OSError(errno.ESRCH, ''), OSError, None])): self.assertTrue(self.p.stop()) + self.assertFalse(self.p.stop()) + self.p.stop_safepoint_reached.clear() + self.assertTrue(self.p.stop()) + with patch.object(Postgresql, '_signal_postmaster_stop', Mock(return_value=(123, None))): + with patch.object(Postgresql, 'is_pid_running', Mock(side_effect=[True, False, False])): + self.assertTrue(self.p.stop()) def test_restart(self): self.p.start = Mock(return_value=False) @@ -778,3 +794,41 @@ class TestPostgresql(unittest.TestCase): self.p.get_server_parameters(config) self.p.set_synchronous_standby('foo') self.p.get_server_parameters(config) + + @patch.object(Postgresql, 'read_pid_file', Mock(return_value={'pid': 'z'})) + def test_get_pid(self): + self.p.get_pid() + + @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + @patch.object(Postgresql, '_signal_postmaster_stop', Mock(return_value=(123, None))) + @patch.object(Postgresql, 'get_pid', Mock(return_value=123)) + @patch('time.sleep', Mock()) + @patch.object(Postgresql, 'is_pid_running') + def test__wait_for_connection_close(self, mock_is_pid_running): + mock_is_pid_running.side_effect = [True, False, False] + self.p.stop_safepoint_reached.clear() + self.p.stop() + + mock_is_pid_running.side_effect = [True, False, False] + self.p.stop_safepoint_reached.clear() + with patch.object(MockCursor, "execute", Mock(side_effect=psycopg2.Error)): + self.p.stop() + + @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + @patch.object(Postgresql, '_signal_postmaster_stop', Mock(return_value=(123, None))) + @patch.object(Postgresql, 'get_pid', Mock(return_value=123)) + @patch.object(Postgresql, 'is_pid_running', Mock(return_value=False)) + @patch('psutil.Process') + def test__wait_for_user_backends_to_close(self, mock_psutil): + child = Mock() + child.cmdline.return_value = ['foo'] + mock_psutil.return_value.children.return_value = [child] + self.p.stop_safepoint_reached.clear() + self.p.stop() + + @patch('os.kill', Mock(side_effect=[OSError(errno.ESRCH, ''), OSError])) + @patch('time.sleep', Mock()) + @patch.object(Postgresql, 'is_pid_running', Mock(side_effect=[True, False])) + def test_terminate_starting_postmaster(self): + self.p.terminate_starting_postmaster(123) + self.p.terminate_starting_postmaster(123) diff --git a/tests/test_watchdog.py b/tests/test_watchdog.py index cfa4d2ca..1ffae833 100644 --- a/tests/test_watchdog.py +++ b/tests/test_watchdog.py @@ -1,12 +1,13 @@ -import unittest -from mock import patch -import platform import ctypes - -from patroni.watchdog import Watchdog import patroni.watchdog.linux as linuxwd - import sys +import unittest + +from mock import patch, Mock, PropertyMock +from patroni.watchdog import Watchdog, WatchdogError +from patroni.watchdog.base import NullWatchdog +from patroni.watchdog.linux import LinuxWatchdogDevice + class MockDevice(object): def __init__(self, fd, filename, flag): @@ -20,41 +21,46 @@ class MockDevice(object): mock_devices = [None] + def mock_open(filename, flag): fd = len(mock_devices) mock_devices.append(MockDevice(fd, filename, flag)) return fd + def mock_ioctl(fd, op, arg=None, mutate_flag=False): assert 0 < fd < len(mock_devices) dev = mock_devices[fd] - sys.stderr.write("Ioctl %d %d %r\n" %( fd, op, arg)) + sys.stderr.write("Ioctl %d %d %r\n" % (fd, op, arg)) if op == linuxwd.WDIOC_GETSUPPORT: sys.stderr.write("Get support\n") - assert(mutate_flag == True) - arg.options = sum(map(linuxwd.WDIOF.get, ['SETTIMEOUT', 'KEEPALIVEPING', 'MAGICCLOSE'])) + assert(mutate_flag is True) + arg.options = sum(map(linuxwd.WDIOF.get, ['SETTIMEOUT', 'KEEPALIVEPING'])) arg.identity = (ctypes.c_ubyte*32)(*map(ord, 'Mock Watchdog')) elif op == linuxwd.WDIOC_GETTIMEOUT: arg.value = dev.timeout elif op == linuxwd.WDIOC_SETTIMEOUT: sys.stderr.write("Set timeout called with %s\n" % arg.value) assert 0 < arg.value < 65535 - dev.timeout = arg.value + dev.timeout = arg.value - 1 else: raise Exception("Unknown op %d", op) return 0 + def mock_write(fd, string): assert 0 < fd < len(mock_devices) assert len(string) == 1 assert mock_devices[fd].open mock_devices[fd].writes.append(string) + def mock_close(fd): assert 0 < fd < len(mock_devices) assert mock_devices[fd].open mock_devices[fd].open = False + @patch('os.open', mock_open) @patch('os.write', mock_write) @patch('os.close', mock_close) @@ -63,18 +69,32 @@ class TestWatchdog(unittest.TestCase): def setUp(self): mock_devices[:] = [None] + @patch('platform.system', Mock(return_value='Linux')) + @patch.object(LinuxWatchdogDevice, 'can_be_disabled', PropertyMock(return_value=True)) + def test_unsafe_timeout_disable_watchdog_and_exit(self): + self.assertRaises(SystemExit, Watchdog({'ttl': 30, 'loop_wait': 15, 'watchdog': {'mode': 'required'}}).activate) + + @patch('platform.system', Mock(return_value='Linux')) + @patch.object(LinuxWatchdogDevice, 'get_timeout', Mock(return_value=16)) + def test_timeout_does_not_ensure_safe_termination(self): + Watchdog({'ttl': 30, 'loop_wait': 15, 'watchdog': {'mode': 'auto'}}).activate() + self.assertEquals(len(mock_devices), 2) + + @patch('platform.system', Mock(return_value='Linux')) + @patch.object(Watchdog, 'is_running', PropertyMock(return_value=False)) + def test_watchdog_not_activated(self): + self.assertRaises(SystemExit, Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'required'}}).activate) + + @patch('platform.system', Mock(return_value='Linux')) def test_basic_operation(self): - if platform.system() != 'Linux': - return - watchdog = Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'required'}}) - watchdog.activate() + self.assertEquals(len(mock_devices), 2) device = mock_devices[-1] self.assertTrue(device.open) - self.assertEquals(device.timeout, 15) + self.assertEquals(device.timeout, 14) watchdog.keepalive() self.assertEquals(len(device.writes), 1) @@ -88,3 +108,56 @@ class TestWatchdog(unittest.TestCase): watchdog.activate() self.assertEquals(len(mock_devices), 1) self.assertFalse(watchdog.is_running) + + def test_parse_mode(self): + with patch('patroni.watchdog.base.logger.warning', new_callable=Mock()) as warning_mock: + watchdog = Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'bad'}}) + self.assertEquals(watchdog.mode, 'off') + warning_mock.assert_called_once() + + @patch('platform.system', Mock(return_value='Unknown')) + def test_unsupported_platform(self): + self.assertRaises(SystemExit, Watchdog, {'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'required'}}) + + def test_exceptions(self): + wd = Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'bad'}}) + wd.impl.close = wd.impl.keepalive = Mock(side_effect=WatchdogError('')) + self.assertIsNone(wd.disable()) + self.assertIsNone(wd.keepalive()) + + +class TestNullWatchdog(unittest.TestCase): + + def test_basics(self): + watchdog = NullWatchdog() + self.assertTrue(watchdog.can_be_disabled) + self.assertRaises(WatchdogError, watchdog.set_timeout, 1) + self.assertEquals(watchdog.describe(), 'NullWatchdog') + self.assertIsInstance(NullWatchdog.from_config({}), NullWatchdog) + + +class TestLinuxWatchdogDevice(unittest.TestCase): + + def setUp(self): + self.impl = LinuxWatchdogDevice.from_config({}) + + @patch('os.open', Mock(return_value=3)) + @patch('os.write', Mock(side_effect=OSError)) + @patch('fcntl.ioctl', Mock(return_value=0)) + def test_basics(self): + self.impl.open() + try: + if self.impl.get_support().has_foo: + self.assertFail() + except Exception as e: + self.assertTrue(isinstance(e, AttributeError)) + self.assertRaises(WatchdogError, self.impl.close) + self.assertRaises(WatchdogError, self.impl.keepalive) + self.assertRaises(WatchdogError, self.impl.set_timeout, -1) + + @patch('os.open', Mock(return_value=3)) + @patch('fcntl.ioctl', Mock(return_value=-1)) + def test__ioctl(self): + self.assertRaises(WatchdogError, self.impl.get_support) + self.impl.open() + self.assertRaises(IOError, self.impl.get_support)