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 7a035b86..fac8a90f 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -82,13 +82,17 @@ 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. - **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/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/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/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index c68bd3d8..82cdff4b 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 @@ -114,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: diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 6a8d1d4b..6fce3018 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): @@ -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' @@ -157,14 +158,16 @@ 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): - 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 @@ -280,7 +283,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 +311,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 +340,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 +513,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,10 +802,11 @@ 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} - 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() @@ -813,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() @@ -939,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 @@ -1047,6 +1058,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)) @@ -1057,9 +1071,23 @@ 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): + """ + 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._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)) + return True + def primary_conninfo(self, member): if not (member and member.conn_url) or member.name == self.name: return None @@ -1287,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 @@ -1339,8 +1323,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') @@ -1348,8 +1334,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/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/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_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_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_ha.py b/tests/test_ha.py index 71d33d25..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): @@ -135,6 +137,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: @@ -245,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 @@ -334,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) @@ -348,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): @@ -796,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() @@ -807,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 a7a477bd..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 @@ -168,9 +170,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'}}, @@ -178,6 +182,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' @@ -213,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) @@ -230,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') @@ -263,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) @@ -498,15 +517,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.config_dir, 'pg_hba.conf')) as f: + lines = f.readlines() + 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() - 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 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 +619,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' @@ -768,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)