Merge branch 'master' of https://github.com/zalando/patroni

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 <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ 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

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 <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ 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 <https://github.com/zalando/patroni/blob/master/docs/dynamic_configuration.rst>`__ 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 <https://github.com/zalando/patroni/blob/master/docs/dynamic_configuration.rst>`__ 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 <https://github.com/zalando/patroni/blob/master/docs/dynamic_configuration.rst>`__ 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**.

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

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':

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)

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))
 

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:

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 ([email protected])
-        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')
 

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':

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

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:

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:

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:

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())

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',

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)

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)

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)