Release 1.6.4 (#1380)

* Bump version
* Update release notes

docs/ENVIRONMENT.rst

@@ -46,7 +46,7 @@ Consul
 -  **PATRONI\_CONSUL\_KEY**: (optional) File with the client key. Can be empty if the key is part of certificate.
 -  **PATRONI\_CONSUL\_DC**: (optional) Datacenter to communicate with. By default the datacenter of the host is used.
 -  **PATRONI\_CONSUL\_CONSISTENCY**: (optional) Select consul consistency mode. Possible values are ``default``, ``consistent``, or ``stale`` (more details in `consul API reference <https://www.consul.io/api/features/consistency.html/>`__)
--  **PATRONI\_CONSUL\_CHECKS**: (optional) list of Consul health checks used for the session. If not specified Consul will use "serfHealth" in additional to the TTL based check created by Patroni. Additional checks, in particular the "serfHealth", may cause the leader lock to expire faster than in `ttl` seconds when the leader instance becomes unavailable.
+-  **PATRONI\_CONSUL\_CHECKS**: (optional) list of Consul health checks used for the session. By default an empty list is used.
 -  **PATRONI\_CONSUL\_REGISTER\_SERVICE**: (optional) whether or not to register a service with the name defined by the scope parameter and the tag master, replica or standby-leader depending on the node's role. Defaults to **false**
 -  **PATRONI\_CONSUL\_SERVICE\_CHECK\_INTERVAL**: (optional) how often to perform health check against registered url
 

docs/SETTINGS.rst

@@ -98,7 +98,7 @@ Most of the parameters are optional, but you have to specify one of the **host**
 -  **key**: (optional) file with the client key. Can be empty if the key is part of **cert**.
 -  **dc**: (optional) Datacenter to communicate with. By default the datacenter of the host is used.
 -  **consistency**: (optional) Select consul consistency mode. Possible values are ``default``, ``consistent``, or ``stale`` (more details in `consul API reference <https://www.consul.io/api/features/consistency.html/>`__)
--  **checks**: (optional) list of Consul health checks used for the session. If not specified Consul will use "serfHealth" in additional to the TTL based check created by Patroni. Additional checks, in particular the "serfHealth", may cause the leader lock to expire faster than in `ttl` seconds when the leader instance becomes unavailable
+-  **checks**: (optional) list of Consul health checks used for the session. By default an empty list is used.
 -  **register\_service**: (optional) whether or not to register a service with the name defined by the scope parameter and the tag master, replica or standby-leader depending on the node's role. Defaults to **false**
 -  **service\_check\_interval**: (optional) how often to perform health check against registered url
 

docs/releases.rst

@@ -3,6 +3,82 @@
 Release notes
 =============
 
+Version 1.6.4
+-------------
+
+**New features**
+
+- Implemented ``--wait`` option for ``patronictl reinit`` (Igor Yanchenko)
+
+  Patronictl will wait for ``reinit`` to finish is the ``--wait`` option is used.
+
+- Further improvements of Windows support (Igor Yanchenko, Alexander Kukushkin)
+
+  1. All shell scripts which are used for integration testing are rewritten in python
+  2. The ``pg_ctl kill`` will be used to stop postgres on non posix systems
+  3. Don't try to use unix-domain sockets
+
+
+**Stability improvements**
+
+- Make sure ``unix_socket_directories`` and ``stats_temp_directory`` exist (Igor)
+
+  Upon the start of Patroni and Postgres make sure that ``unix_socket_directories`` and ``stats_temp_directory`` exist or try to create them. Patroni will exit if failed to create them.
+
+- Make sure ``postgresql.pgpass`` is located in the place where Patroni has write access (Igor)
+
+  In case if it doesn't have a write access Patroni will exit with exception.
+
+- Disable Consul ``serfHealth`` check by default (Kostiantyn Nemchenko)
+
+  Even in case of little network problems the failing ``serfHealth`` leads to invalidation of all sessions associated with the node. Therefore, the leader key is lost much earlier than ``ttl`` which causes unwanted restarts of replicas and maybe demotion of the primary.
+
+- Configure tcp keepalives for connections to K8s API (Alexander)
+
+  In case if we get nothing from the socket after TTL seconds it can be considered dead.
+
+- Avoid logging of passwords on user creation (Alexander)
+
+  If the password is rejected or logging is configured to verbose or not configured at all it might happen that the password is written into postgres logs. In order to avoid it Patroni will change ``log_statement``, ``log_min_duration_statement``, and ``log_min_error_statement`` to some safe values before doing the attempt to create/update user.
+
+
+**Bugfixes**
+
+- Use ``restore_command`` from the ``standby_cluster`` config on cascading replicas (Alexander)
+
+  The ``standby_leader`` was already doing it from the beginning the feature existed. Not doing the same on replicas might prevent them from catching up with standby leader.
+
+- Update timeline reported by the standby cluster (Alexander)
+
+  In case of timeline switch the standby cluster was correctly replicating from the primary but ``patronictl`` was reporting the old timeline.
+
+- Allow certain recovery parameters be defined in the custom_conf (Alexander)
+
+  When doing validation of recovery parameters on replica Patroni will skip ``archive_cleanup_command``, ``promote_trigger_file``, ``recovery_end_command``, ``recovery_min_apply_delay``, and ``restore_command`` if they are not defined in the patroni config but in files other than ``postgresql.auto.conf`` or ``postgresql.conf``.
+
+- Improve handling of postgresql parameters with period in its name (Alexander)
+
+  Such parameters could be defined by extensions where the unit is not necessarily a string. Changing the value might require a restart (for example ``pg_stat_statements.max``).
+
+- Improve exception handling during shutdown (Alexander)
+
+  During shutdown Patroni is trying to update its status in the DCS. If the DCS is inaccessible an exception might be raised. Lack of exception handling was preventing logger thread from stopping.
+
+
+Version 1.6.3
+-------------
+
+**Bugfixes**
+
+- Don't expose password when running ``pg_rewind`` (Alexander Kukushkin)
+
+  Bug was introduced in the `#1301 <https://github.com/zalando/patroni/pull/1301>`__
+
+- Apply connection parameters specified in the ``postgresql.authentication`` to ``pg_basebackup`` and custom replica creation methods (Alexander)
+
+  They were relying on url-like connection string and therefore parameters never applied.
+
+
 Version 1.6.2
 -------------
 

features/archive-restore.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+import os
+import argparse
+import shutil
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dirname", required=True)
+    parser.add_argument("--pathname", required=True)
+    parser.add_argument("--filename", required=True)
+    parser.add_argument("--mode", required=True, choices=("archive", "restore"))
+    args, _ = parser.parse_known_args()
+
+    full_filename = os.path.join(args.dirname, args.filename)
+    if args.mode == "archive":
+        if not os.path.isdir(args.dirname):
+            os.makedirs(args.dirname)
+        if not os.path.exists(full_filename):
+            shutil.copy(args.pathname, full_filename)
+    else:
+        shutil.copy(full_filename, args.pathname)

features/backup_create.py

@@ -0,0 +1,14 @@
+#!/usr/bin/env python
+import argparse
+import subprocess
+import sys
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--datadir", required=True)
+    parser.add_argument("--dbname", required=True)
+    parser.add_argument("--walmethod", required=True, choices=("fetch", "stream", "none"))
+    args, _ = parser.parse_known_args()
+
+    walmethod = ["-X", args.walmethod] if args.walmethod != "none" else []
+    sys.exit(subprocess.call(["pg_basebackup", "-D", args.datadir, "-c", "fast", "-d", args.dbname] + walmethod))

features/backup_create.sh

@@ -1,22 +0,0 @@
-#!/bin/bash
-
-while getopts ":-:" optchar; do
-    [[ "${optchar}" == "-" ]] || continue
-    case "${OPTARG}" in
-        datadir=* )
-            PGDATA=${OPTARG#*=}
-            ;;
-        dbname=* )
-            DBNAME=${OPTARG#*=}
-            ;;
-        walmethod=* )
-            WALMETHOD=${OPTARG#*=}
-            ;;
-    esac
-done
-
-[[ -z $PGDATA || -z $DBNAME || -z $WALMETHOD ]] && exit 1
-
-[[ $WALMETHOD != "none" ]] && WALMETHOD="-X $WALMETHOD" || WALMETHOD=""
-
-exec pg_basebackup -D $PGDATA $WALMETHOD -c fast -d $DBNAME

features/backup_restore.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python
+import argparse
+import shutil
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--datadir", required=True)
+    parser.add_argument("--sourcedir", required=True)
+    args, _ = parser.parse_known_args()
+
+    shutil.copytree(args.sourcedir, args.datadir)

features/backup_restore.sh

@@ -1,21 +0,0 @@
-#!/bin/bash
-
-set -x
-
-while getopts ":-:" optchar; do
-    [[ "${optchar}" == "-" ]] || continue
-    case "${OPTARG}" in
-        datadir=* )
-            PGDATA=${OPTARG#*=}
-            ;;
-        sourcedir=* )
-            SOURCE=${OPTARG#*=}
-            ;;
-    esac
-done
-
-[[ -z $PGDATA || -z $SOURCE ]] && exit 1
-
-mkdir -p $(dirname $PGDATA)
-
-exec cp -af $SOURCE $PGDATA

features/callback.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python
+import os
+import psycopg2
+import sys
+
+
+if __name__ == '__main__':
+    if not (len(sys.argv) >= 3 and sys.argv[3] == "master"):
+        sys.exit(1)
+
+    os.environ['PGPASSWORD'] = 'zalando'
+    connection = psycopg2.connect(host='127.0.0.1', port=sys.argv[1], user='postgres')
+    cursor = connection.cursor()
+    cursor.execute("SELECT slot_name FROM pg_replication_slots WHERE slot_type = 'logical'")
+
+    with open("data/postgres0/label", "w") as label:
+        label.write(next(iter(cursor.fetchone()), ""))

features/callback.sh

@@ -1,5 +0,0 @@
-#!/bin/bash
-
-[[ "$3" == "master" ]] || exit
-
-PGPASSWORD=zalando psql -h localhost -U postgres -p $1 -w -tAc "SELECT slot_name FROM pg_replication_slots WHERE slot_type = 'logical'" >> data/postgres0/label

features/callback2.py

@@ -0,0 +1,5 @@
+#!/usr/bin/env python
+
+import sys
+with open("data/{0}/{0}_cb.log".format(sys.argv[1]), "a+") as log:
+    log.write(" ".join(sys.argv[-3:]) + "\n")

features/environment.py

@@ -171,7 +171,8 @@ class PatroniController(AbstractController):
 
         config['name'] = name
         config['postgresql']['data_dir'] = self._data_dir
-        config['postgresql']['use_unix_socket'] = True
+        config['postgresql']['use_unix_socket'] = os.name != 'nt'  # windows doesn't yet support unix-domain sockets
+        config['postgresql']['pgpass'] = os.path.join(tempfile.gettempdir(), 'pgpass_' + name)
         config['postgresql']['parameters'].update({
             'logging_collector': 'on', 'log_destination': 'csvlog', 'log_directory': self._output_dir,
             'log_filename': name + '.log', 'log_statement': 'all', 'log_min_messages': 'debug1',
@@ -258,10 +259,10 @@ class PatroniController(AbstractController):
     def backup_source(self):
         return 'postgres://{username}:{password}@{host}:{port}/{database}'.format(**self._replication)
 
-    def backup(self, dest='data/basebackup'):
-        subprocess.call([PatroniPoolController.BACKUP_SCRIPT, '--walmethod=none',
-                         '--datadir=' + os.path.join(self._work_directory, dest),
-                         '--dbname=' + self.backup_source])
+    def backup(self, dest=os.path.join('data', 'basebackup')):
+        subprocess.call(PatroniPoolController.BACKUP_SCRIPT + ['--walmethod=none',
+                        '--datadir=' + os.path.join(self._work_directory, dest),
+                        '--dbname=' + self.backup_source])
 
 
 class ProcessHang(object):
@@ -532,7 +533,8 @@ class ExhibitorController(ZooKeeperController):
 
 class PatroniPoolController(object):
 
-    BACKUP_SCRIPT = 'features/backup_create.sh'
+    BACKUP_SCRIPT = [sys.executable, 'features/backup_create.py']
+    ARCHIVE_RESTORE_SCRIPT = ' '.join((sys.executable, os.path.abspath('features/archive-restore.py')))
 
     def __init__(self, context):
         self._context = context
@@ -593,7 +595,7 @@ class PatroniPoolController(object):
             'bootstrap': {
                 'method': 'pg_basebackup',
                 'pg_basebackup': {
-                    'command': self.BACKUP_SCRIPT + ' --walmethod=stream --dbname=' + f.backup_source
+                    'command': " ".join(self.BACKUP_SCRIPT) + ' --walmethod=stream --dbname=' + f.backup_source
                 },
                 'dcs': {
                     'postgresql': {
@@ -606,8 +608,9 @@ class PatroniPoolController(object):
             'postgresql': {
                 'parameters': {
                     'archive_mode': 'on',
-                    'archive_command': 'mkdir -p {0} && test ! -f {0}/%f && cp %p {0}/%f'.format(
-                            os.path.join(self.patroni_path, 'data', 'wal_archive'))
+                    'archive_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode archive ' +
+                                        '--dirname {} --filename %f --pathname %p').format(
+                                        os.path.join(self.patroni_path, 'data', 'wal_archive'))
                 },
                 'authentication': {
                     'superuser': {'password': 'zalando1'},
@@ -623,12 +626,14 @@ class PatroniPoolController(object):
             'bootstrap': {
                 'method': 'backup_restore',
                 'backup_restore': {
-                    'command': 'features/backup_restore.sh --sourcedir=' + os.path.join(self.patroni_path,
-                                                                                        'data', 'basebackup'),
+                    'command': (sys.executable + ' features/backup_restore.py --sourcedir=' +
+                                os.path.join(self.patroni_path, 'data', 'basebackup')),
                     'recovery_conf': {
                         'recovery_target_action': 'promote',
                         'recovery_target_timeline': 'latest',
-                        'restore_command': 'cp {0}/data/wal_archive/%f %p'.format(self.patroni_path)
+                        'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' +
+                                            '--dirname {} --filename %f --pathname %p').format(
+                                            os.path.join(self.patroni_path, 'data', 'wal_archive'))
                     }
                 }
             },

features/standby_cluster.feature

@@ -30,7 +30,7 @@ Feature: standby cluster
     When I issue a GET request to http://127.0.0.1:8009/standby_leader
     Then I receive a response code 200
     And I receive a response role standby_leader
-    And there is a postgres1_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres1 data directory
+    And there is a postgres1_cb.log with "on_role_change standby_leader batman1" in postgres1 data directory
     When I start postgres2 in a cluster batman1
     Then postgres2 role is the replica after 24 seconds
     And table foo is present on postgres2 after 20 seconds

features/steps/cascading_replication.py

@@ -13,7 +13,7 @@ def start_patroni_with_a_name_value_tag(context, name, tag_name, tag_value):
 def check_label(context, label, content, name):
     label = context.pctl.read_label(name, label)
     label = label.replace('\n', '\\n')
-    assert label == content, "{0} is not equal to {1}".format(label, content)
+    assert content in label, "{0} doesn't contain {1}".format(label, content)
 
 
 @step('I create label with "{content:w}" in {name:w} data directory')

features/steps/standby_cluster.py

@@ -1,4 +1,5 @@
 import os
+import sys
 import time
 
 from behave import step
@@ -9,7 +10,7 @@ SELECT * FROM pg_catalog.pg_stat_replication
 WHERE application_name = '{0}'
 """
 
-callback = "bash -c 'echo \"${*: -3:1} ${*: -2:1} ${*: -1:1}\" >> data/$1/$1_cb.log' -- "
+callback = sys.executable + " features/callback2.py "
 
 
 @step('I start {name:w} with callback configured')
@@ -17,7 +18,7 @@ def start_patroni_with_callbacks(context, name):
     return context.pctl.start(name, custom_config={
         "postgresql": {
             "callbacks": {
-                "on_role_change": "features/callback.sh"
+                "on_role_change": sys.executable + " features/callback.py"
             }
         }
     })
@@ -30,8 +31,8 @@ def start_patroni(context, name, cluster_name):
         "postgresql": {
             "callbacks": {c: callback + name for c in ('on_start', 'on_stop', 'on_restart', 'on_role_change')},
             "backup_restore": {
-                "command": "features/backup_restore.sh --sourcedir=" + os.path.join(context.pctl.patroni_path,
-                                                                                    'data', 'basebackup')}
+                "command": (sys.executable + " features/backup_restore.py --sourcedir=" +
+                            os.path.join(context.pctl.patroni_path, 'data', 'basebackup'))}
         }
     })
 

patroni/__init__.py

@@ -159,7 +159,10 @@ class Patroni(object):
             self.api.shutdown()
         except Exception:
             logger.exception('Exception during RestApi.shutdown')
-        self.ha.shutdown()
+        try:
+            self.ha.shutdown()
+        except Exception:
+            logger.exception('Exception during Ha.shutdown')
         self.logger.shutdown()
 
 

patroni/ctl.py

@@ -49,11 +49,11 @@ class PatroniCtlException(ClickException):
 def parse_dcs(dcs):
     if dcs is None:
         return None
+    elif '//' not in dcs:
+        dcs = '//' + dcs
 
     parsed = urlparse(dcs)
     scheme = parsed.scheme
-    if scheme == '' and parsed.netloc == '':
-        parsed = urlparse('//' + dcs)
     port = int(parsed.port) if parsed.port else None
 
     if scheme == '':
@@ -558,20 +558,39 @@ def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, vers
 @click.argument('cluster_name')
 @click.argument('member_names', nargs=-1)
 @option_force
+@click.option('--wait', help='Wait until reinitialization completes', is_flag=True)
 @click.pass_obj
-def reinit(obj, cluster_name, member_names, force):
+def reinit(obj, cluster_name, member_names, force, wait):
     cluster = get_dcs(obj, cluster_name).get_cluster()
     members = get_members(cluster, cluster_name, member_names, None, force, 'reinitialize')
 
+    wait_on_members = []
     for member in members:
         body = {'force': force}
         while True:
             r = request_patroni(member, 'post', 'reinitialize', body)
-            if not check_response(r, member.name, 'reinitialize') and r.data.endswith(b' already in progress') \
+            started = check_response(r, member.name, 'reinitialize')
+            if not started and r.data.endswith(b' already in progress') \
                     and not force and click.confirm('Do you want to cancel it and reinitialize anyway?'):
                 body['force'] = True
                 continue
             break
+        if started and wait:
+            wait_on_members.append(member)
+
+    last_display = []
+    while wait_on_members:
+        if wait_on_members != last_display:
+            click.echo('Waiting for reinitialize to complete on: {0}'.format(
+                ", ".join(member.name for member in wait_on_members))
+            )
+            last_display[:] = wait_on_members
+        time.sleep(2)
+        for member in wait_on_members:
+            data = json.loads(request_patroni(member, 'get', 'patroni').data.decode('utf-8'))
+            if data.get('state') != 'creating replica':
+                click.echo('Reinitialize is completed on: {0}'.format(member.name))
+                wait_on_members.remove(member)
 
 
 def _do_failover_or_switchover(obj, action, cluster_name, master, candidate, force, scheduled=None):

patroni/dcs/__init__.py

@@ -167,7 +167,7 @@ class Member(namedtuple('Member', 'index,name,session,data')):
 
         # apply any remaining authentication parameters
         if auth and isinstance(auth, dict):
-            ret.update(auth)
+            ret.update({k: v for k, v in auth.items() if v is not None})
             if 'username' in auth:
                 ret['user'] = ret.pop('username')
         return ret
@@ -240,21 +240,25 @@ class Leader(namedtuple('Leader', 'index,session,member')):
     def conn_url(self):
         return self.member.conn_url
 
+    @property
+    def data(self):
+        return self.member.data
+
     @property
     def timeline(self):
-        return self.member.data.get('timeline')
+        return self.data.get('timeline')
 
     @property
     def checkpoint_after_promote(self):
         """
         >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote
         """
-        version = self.member.data.get('version')
+        version = self.data.get('version')
         if version:
             try:
                 # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false
                 if tuple(map(int, version.split('.'))) > (1, 5, 6):
-                    return self.member.data['role'] == 'master' and 'checkpoint_after_promote' not in self.member.data
+                    return self.data['role'] == 'master' and 'checkpoint_after_promote' not in self.data
             except Exception:
                 logger.debug('Failed to parse Patroni version %s', version)
 

patroni/dcs/consul.py

@@ -217,7 +217,7 @@ class Consul(AbstractDCS):
         self.set_retry_timeout(config['retry_timeout'])
         self.set_ttl(config.get('ttl') or 30)
         self._last_session_refresh = 0
-        self.__session_checks = config.get('checks')
+        self.__session_checks = config.get('checks', [])
         self._register_service = config.get('register_service', False)
         if self._register_service:
             self._service_name = service_name_from_scope_name(self._scope)

patroni/dcs/etcd.py

@@ -73,8 +73,8 @@ class DnsCachingResolver(Thread):
     def _do_resolve(host, port):
         try:
             return socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM, socket.IPPROTO_TCP)
-        except socket.gaierror:
-            logger.warning('failed to resolve host %s', host)
+        except Exception as e:
+            logger.warning('failed to resolve host %s: %s', host, e)
             return []
 
 

patroni/dcs/kubernetes.py

@@ -41,8 +41,22 @@ class CoreV1ApiProxy(object):
         self._request_timeout = None
         self._use_endpoints = use_endpoints
 
-    def set_timeout(self, timeout):
-        self._request_timeout = (1, timeout / 3.0)
+    def configure_timeouts(self, loop_wait, retry_timeout, ttl):
+        # Normally every loop_wait seconds we should have receive something from the socket.
+        # If we didn't received anything after the loop_wait + retry_timeout it is a time
+        # to start worrying (send keepalive messages). Finally, the connection should be
+        # considered as dead if we received nothing from the socket after the ttl seconds.
+        cnt = 3
+        idle = int(loop_wait + retry_timeout)
+        intvl = max(1, int(float(ttl - idle) / cnt))
+        self._api.api_client.rest_client.pool_manager.connection_pool_kw['socket_options'] = [
+            (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
+            (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, idle),
+            (socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, intvl),
+            (socket.IPPROTO_TCP, socket.TCP_KEEPCNT, cnt),
+            (socket.IPPROTO_TCP, 18, int(ttl * 1000))  # TCP_USER_TIMEOUT
+        ]
+        self._request_timeout = (1, retry_timeout / 3.0)
 
     def __getattr__(self, func):
         if func.endswith('_kind'):
@@ -210,8 +224,7 @@ class Kubernetes(AbstractDCS):
             self.__subsets = [k8s_client.V1EndpointSubset(addresses=addresses, ports=ports)]
             self._should_create_config_service = True
         self._api = CoreV1ApiProxy(use_endpoints)
-        self.set_retry_timeout(config['retry_timeout'])
-        self.set_ttl(config.get('ttl') or 30)
+        self.reload_config(config)
         self._leader_observed_record = {}
         self._leader_observed_time = None
         self._leader_resource_version = None
@@ -250,7 +263,10 @@ class Kubernetes(AbstractDCS):
 
     def set_retry_timeout(self, retry_timeout):
         self._retry.deadline = retry_timeout
-        self._api.set_timeout(retry_timeout)
+
+    def reload_config(self, config):
+        super(Kubernetes, self).reload_config(config)
+        self._api.configure_timeouts(self.loop_wait, self._retry.deadline, self.ttl)
 
     @staticmethod
     def member(pod):

patroni/ha.py

@@ -191,10 +191,20 @@ class Ha(object):
             if self._async_executor.scheduled_action in (None, 'promote') \
                     and data['state'] in ['running', 'restarting', 'starting']:
                 try:
-                    timeline, wal_position = self.state_handler.timeline_wal_position()
+                    timeline, wal_position, pg_control_timeline = self.state_handler.timeline_wal_position()
                     data['xlog_location'] = wal_position
                     if not timeline:
-                        timeline = self.state_handler.replica_cached_timeline(self._leader_timeline)
+                        # So far the only way to get the current timeline on the standby is from
+                        # the replication connection. In order to avoid opening the replication
+                        # connection on every iteration of HA loop we will do it only when noticed
+                        # that the timeline on the primary has changed.
+                        # Unfortunately such optimization isn't possible on the standby_leader,
+                        # therefore we will get the timeline from pg_control, either by calling
+                        # pg_control_checkpoint() on 9.6+ or by parsing the output of pg_controldata.
+                        if self.state_handler.role == 'standby_leader':
+                            timeline = pg_control_timeline or self.state_handler.pg_control_timeline()
+                        else:
+                            timeline = self.state_handler.replica_cached_timeline(timeline)
                     if timeline:
                         data['timeline'] = timeline
                 except Exception:
@@ -342,14 +352,26 @@ class Ha(object):
     def _get_node_to_follow(self, cluster):
         # determine the node to follow. If replicatefrom tag is set,
         # try to follow the node mentioned there, otherwise, follow the leader.
-        if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)):
+        standby_config = self.get_standby_cluster_config()
+        is_standby_cluster = _is_standby_cluster(standby_config)
+        if is_standby_cluster and (self.cluster.is_unlocked() or self.has_lock(False)):
             node_to_follow = self.get_remote_master()
         elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
             node_to_follow = cluster.get_member(self.patroni.replicatefrom)
         else:
             node_to_follow = cluster.leader
 
-        return node_to_follow if node_to_follow and node_to_follow.name != self.state_handler.name else None
+        node_to_follow = node_to_follow if node_to_follow and node_to_follow.name != self.state_handler.name else None
+
+        if node_to_follow and not isinstance(node_to_follow, RemoteMember):
+            # we are going to abuse Member.data to pass following parameters
+            params = ('restore_command', 'archive_cleanup_command')
+            for param in params:  # It is highly unlikely to happen, but we want to protect from the case
+                node_to_follow.data.pop(param, None)  # when above-mentioned params came from outside.
+            if is_standby_cluster:
+                node_to_follow.data.update({p: standby_config[p] for p in params if standby_config.get(p)})
+
+        return node_to_follow
 
     def follow(self, demote_reason, follow_reason, refresh=True):
         if refresh:
@@ -585,7 +607,7 @@ class Ha(object):
         """This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
 
         # We don't call `last_operation()` here because it returns a string
-        _, my_wal_position = self.state_handler.timeline_wal_position()
+        _, my_wal_position, _ = self.state_handler.timeline_wal_position()
         if check_replication_lag and self.is_lagging(my_wal_position):
             logger.info('My wal position exceeds maximum replication lag')
             return False  # Too far behind last reported wal position on master
@@ -1268,7 +1290,8 @@ class Ha(object):
                 data_sysid = self.state_handler.sysid
                 if not self.sysid_valid(data_sysid):
                     # data directory is not empty, but no valid sysid, cluster must be broken, suggest reinit
-                    return "data dir for the cluster is not empty, but system ID is invalid; consider doing reinitalize"
+                    return ("data dir for the cluster is not empty, but system ID is invalid; consider doing"
+                            "reinitialize")
 
                 if self.sysid_valid(self.cluster.initialize):
                     if self.cluster.initialize != data_sysid:

patroni/postgresql/__init__.py

@@ -60,6 +60,7 @@ class Postgresql(object):
         self._pending_restart = False
         self._connection = Connection()
         self.config = ConfigHandler(self, config)
+        self.config.check_directories()
 
         self._bin_dir = config.get('bin_dir') or ''
         self.bootstrap = Bootstrap(self)
@@ -133,6 +134,8 @@ class Postgresql(object):
 
     @property
     def cluster_info_query(self):
+        pg_control_timeline = 'timeline_id FROM pg_catalog.pg_control_checkpoint()' \
+                if self._major_version >= 90600 and self.role == 'standby_leader' else '0'
         return ("SELECT CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 "
                 "ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name("
                 "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, "
@@ -141,7 +144,7 @@ class Postgresql(object):
                 "pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint,"
                 " pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint)"
                 "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint "
-                "END").format(self.wal_name, self.lsn_name)
+                "END, {2}").format(self.wal_name, self.lsn_name, pg_control_timeline)
 
     def _version_file_exists(self):
         return not self.data_directory_empty() and os.path.isfile(self._version_file)
@@ -288,7 +291,7 @@ class Postgresql(object):
         if not self._cluster_info_state:
             try:
                 result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone()
-                self._cluster_info_state = dict(zip(['timeline', 'wal_position'], result))
+                self._cluster_info_state = dict(zip(['timeline', 'wal_position', 'pg_control_timeline'], result))
             except RetryFailedError as e:  # SELECT failed two times
                 self._cluster_info_state = {'error': str(e)}
                 if not self.is_starting() and self.pg_isready() == STATE_REJECT:
@@ -302,6 +305,12 @@ class Postgresql(object):
     def is_leader(self):
         return bool(self._cluster_info_state_get('timeline'))
 
+    def pg_control_timeline(self):
+        try:
+            return int(self.controldata().get("Latest checkpoint's TimeLineID"))
+        except (TypeError, ValueError):
+            logger.exception('Failed to parse timeline from pg_controldata output')
+
     def is_running(self):
         """Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process
         is running updates the cached process based on pid file."""
@@ -407,6 +416,7 @@ class Postgresql(object):
         self._pending_restart = False
 
         configuration = self.config.effective_configuration
+        self.config.check_directories()
         self.config.write_postgresql_conf(configuration)
         self.config.resolve_connection_addresses()
         self.config.replace_pg_hba()
@@ -506,7 +516,7 @@ class Postgresql(object):
             self.set_state('stopping')
 
         # Send signal to postmaster to stop
-        success = postmaster.signal_stop(mode)
+        success = postmaster.signal_stop(mode, self.pgcommand('pg_ctl'))
         if success is not None:
             if success and on_safepoint:
                 on_safepoint()
@@ -523,11 +533,10 @@ class Postgresql(object):
 
         return True, True
 
-    @staticmethod
-    def terminate_starting_postmaster(postmaster):
+    def terminate_starting_postmaster(self, postmaster):
         """Terminates a postmaster that has not yet opened ports or possibly even written a pid file. Blocks
         until the process goes away."""
-        postmaster.signal_stop('immediate')
+        postmaster.signal_stop('immediate', self.pgcommand('pg_ctl'))
         postmaster.wait()
 
     def _wait_for_connection_close(self, postmaster):
@@ -733,11 +742,13 @@ class Postgresql(object):
         # This method could be called from different threads (simultaneously with some other `_query` calls).
         # If it is called not from main thread we will create a new cursor to execute statement.
         if current_thread().ident == self.__thread_ident:
-            return self._cluster_info_state_get('timeline'), self._cluster_info_state_get('wal_position')
+            return (self._cluster_info_state_get('timeline'),
+                    self._cluster_info_state_get('wal_position'),
+                    self._cluster_info_state_get('pg_control_timeline'))
 
         with self.connection().cursor() as cursor:
             cursor.execute(self.cluster_info_query)
-            return cursor.fetchone()[:2]
+            return cursor.fetchone()[:3]
 
     def postmaster_start_time(self):
         try:
@@ -815,7 +826,7 @@ class Postgresql(object):
             if state != 'streaming' or not member or member.tags.get('nosync', False):
                 continue
             if sync_state == 'sync':
-                return app_name, True
+                return member.name, True
             if sync_state == 'potential' and app_name == current:
                 # Prefer current even if not the best one any more to avoid indecisivness and spurious swaps.
                 return cluster.sync.sync_standby, False

patroni/postgresql/bootstrap.py

@@ -5,9 +5,8 @@ import tempfile
 import time
 
 from patroni.dcs import RemoteMember
-from patroni.utils import deep_compare, uri
+from patroni.utils import deep_compare
 from six import string_types
-from six.moves.urllib.parse import quote_plus
 
 logger = logging.getLogger(__name__)
 
@@ -123,19 +122,13 @@ class Bootstrap(object):
         cmd = config.get('post_bootstrap') or config.get('post_init')
         if cmd:
             r = self._postgresql.config.local_connect_kwargs
-
-            if 'host' in r:
-                # '/tmp' => '%2Ftmp' for unix socket path
-                host = quote_plus(r['host']) if r['host'].startswith('/') else r['host']
-            else:
-                host = ''
-
+            connstring = self._postgresql.config.format_dsn(r, True)
+            if 'host' not in r:
                 # https://www.postgresql.org/docs/current/static/libpq-pgpass.html
                 # A host name of localhost matches both TCP (host name localhost) and Unix domain socket
                 # (pghost empty or the default socket directory) connections coming from the local machine.
                 r['host'] = 'localhost'  # set it to localhost to write into pgpass
 
-            connstring = uri('postgres', (host, r['port']), r['database'], r.get('user'))
             env = self._postgresql.config.write_pgpass(r) if 'password' in r else None
 
             try:
@@ -168,9 +161,9 @@ class Bootstrap(object):
 
         if clone_member and clone_member.conn_url:
             r = clone_member.conn_kwargs(self._postgresql.config.replication)
-            connstring = uri('postgres', (r['host'], r['port']), r['database'], r['user'])
             # add the credentials to connect to the replica origin to pgpass.
             env = self._postgresql.config.write_pgpass(r)
+            connstring = self._postgresql.config.format_dsn(r, True)
         else:
             connstring = ''
             env = os.environ.copy()
@@ -317,7 +310,15 @@ BEGIN
         CREATE ROLE "{0}" WITH {1};
     END IF;
 END;$$""".format(name, ' '.join(options))
-        self._postgresql.query(sql, *params)
+        self._postgresql.query('SET log_statement TO none')
+        self._postgresql.query('SET log_min_duration_statement TO -1')
+        self._postgresql.query("SET log_min_error_statement TO 'log'")
+        try:
+            self._postgresql.query(sql, *params)
+        finally:
+            self._postgresql.query('RESET log_min_error_statement')
+            self._postgresql.query('RESET log_min_duration_statement')
+            self._postgresql.query('RESET log_statement')
 
     def post_bootstrap(self, config, task):
         try:

patroni/postgresql/config.py

@@ -6,11 +6,13 @@ import socket
 import stat
 import time
 
+from patroni.exceptions import PatroniException
 from six.moves.urllib_parse import urlparse, parse_qsl, unquote
 from urllib3.response import HTTPHeaderDict
 
 from ..dcs import slot_name_from_member_name, RemoteMember
-from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri
+from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, \
+        validate_directory, is_subpath
 
 logger = logging.getLogger(__name__)
 
@@ -333,7 +335,10 @@ class ConfigHandler(object):
         self._standby_signal = os.path.join(postgresql.data_dir, 'standby.signal')
         self._auto_conf = os.path.join(postgresql.data_dir, 'postgresql.auto.conf')
         self._auto_conf_mtime = None
-        self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
+        self._pgpass = os.path.abspath(config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass'))
+        if os.path.exists(self._pgpass) and not os.path.isfile(self._pgpass):
+            raise PatroniException("'{}' exists and it's not a file, check your `postgresql.pgpass` configuration"
+                                   .format(self._pgpass))
         self._passfile = None
         self._passfile_mtime = None
         self._synchronous_standby_names = None
@@ -347,6 +352,21 @@ class ConfigHandler(object):
         self._server_parameters = self.get_server_parameters(self._config)
         self._adjust_recovery_parameters()
 
+    def try_to_create_dir(self, d, msg):
+        d = os.path.join(self._postgresql._data_dir, d)
+        if (not is_subpath(self._postgresql._data_dir, d) or not self._postgresql.data_directory_empty()):
+            validate_directory(d, msg)
+
+    def check_directories(self):
+        if "unix_socket_directories" in self._server_parameters:
+            for d in self._server_parameters["unix_socket_directories"].split(","):
+                self.try_to_create_dir(d.strip(), "'{}' is defined in unix_socket_directories, {}")
+        if "stats_temp_directory" in self._server_parameters:
+            self.try_to_create_dir(self._server_parameters["stats_temp_directory"],
+                                   "'{}' is defined in stats_temp_directory, {}")
+        self.try_to_create_dir(os.path.dirname(self._pgpass),
+                               "'{}' is defined in `postgresql.pgpass`, {}")
+
     @property
     def _configuration_to_save(self):
         configuration = [os.path.basename(self._postgresql_conf)]
@@ -436,7 +456,7 @@ class ConfigHandler(object):
         # when we are doing custom bootstrap we assume that we don't know superuser password
         # and in order to be able to change it, we are opening trust access from a certain address
         if self._postgresql.bootstrap.running_custom_bootstrap:
-            addresses = {'': 'local'}
+            addresses = {} if os.name == 'nt' else {'': 'local'}  # windows doesn't yet support unix-domain sockets
             if 'host' in self.local_replication_address and not self.local_replication_address['host'].startswith('/'):
                 addresses.update({sa[0] + '/32': 'host' for _, _, _, _, sa in socket.getaddrinfo(
                                   self.local_replication_address['host'], self.local_replication_address['port'],
@@ -480,17 +500,22 @@ class ConfigHandler(object):
 
     def format_dsn(self, params, include_dbname=False):
         # A list of keywords that can be found in a conninfo string. Follows what is acceptable by libpq
-        keywords = ('user', 'passfile' if params.get('passfile') else 'password', 'host', 'port', 'sslmode',
+        keywords = ('dbname', 'user', 'passfile' if params.get('passfile') else 'password', 'host', 'port', 'sslmode',
                     'sslcompression', 'sslcert', 'sslkey', 'sslrootcert', 'sslcrl', 'application_name', 'krbsrvname')
         if include_dbname:
             params = params.copy()
             params['dbname'] = params.get('database') or self._postgresql.database
-            keywords = ('dbname',) + keywords
+            # we are abusing information about the necessity of dbname
+            # dsn should contain passfile or password only if there is no dbname in it (it is used in recovery.conf)
+            skip = {'passfile', 'password'}
+        else:
+            skip = {'dbname'}
 
         def escape(value):
             return re.sub(r'([\'\\ ])', r'\\\1', str(value))
 
-        return ' '.join('{0}={1}'.format(kw, escape(params[kw])) for kw in keywords if params.get(kw) is not None)
+        return ' '.join('{0}={1}'.format(kw, escape(params[kw])) for kw in keywords
+                        if kw not in skip and params.get(kw) is not None)
 
     def _write_recovery_params(self, fd, recovery_params):
         for name, value in sorted(recovery_params.items()):
@@ -516,15 +541,16 @@ class ConfigHandler(object):
         is_remote_master = isinstance(member, RemoteMember)
         primary_conninfo = self.primary_conninfo_params(member)
         if primary_conninfo:
+            use_slots = self.get('use_slots', True) and self._postgresql.major_version >= 90400
+            if use_slots and not (is_remote_master and member.no_replication_slot):
+                primary_slot_name = member.primary_slot_name if is_remote_master else self._postgresql.name
+                recovery_params['primary_slot_name'] = slot_name_from_member_name(primary_slot_name)
             recovery_params['primary_conninfo'] = primary_conninfo
-            if self.get('use_slots', True) and self._postgresql.major_version >= 90400 \
-                    and not (is_remote_master and member.no_replication_slot):
-                recovery_params['primary_slot_name'] = member.primary_slot_name if is_remote_master \
-                        else slot_name_from_member_name(self._postgresql.name)
 
-        if is_remote_master:  # standby_cluster config might have different parameters, we want to override them
-            recovery_params.update({p: member.data.get(p) for p in ('restore_command', 'recovery_min_apply_delay',
-                                                                    'archive_cleanup_command') if member.data.get(p)})
+        # standby_cluster config might have different parameters, we want to override them
+        standby_cluster_params = ['restore_command', 'archive_cleanup_command']\
+            + (['recovery_min_apply_delay'] if is_remote_master else [])
+        recovery_params.update({p: member.data.get(p) for p in standby_cluster_params if member and member.data.get(p)})
         return recovery_params
 
     def recovery_conf_exists(self):
@@ -559,7 +585,7 @@ class ConfigHandler(object):
 
         try:
             values = self._get_pg_settings(self._recovery_parameters_to_compare).values()
-            values = {p[0]: [p[1], p[4] == 'postmaster'] for p in values}
+            values = {p[0]: [p[1], p[4] == 'postmaster', p[5]] for p in values}
             self._postgresql_conf_mtime = pg_conf_mtime
             self._auto_conf_mtime = auto_conf_mtime
             self._postmaster_ctime = postmaster_ctime
@@ -668,6 +694,12 @@ class ConfigHandler(object):
 
         wanted_recovery_params = self.build_recovery_params(member)
         for param, value in self._current_recovery_params.items():
+            # Skip certain parameters defined in the included postgres config files
+            # if we know that they are not specified in the patroni configuration.
+            if len(value) > 2 and value[2] not in (self._postgresql_conf, self._auto_conf) and \
+                    param in ('archive_cleanup_command', 'promote_trigger_file', 'recovery_end_command',
+                              'recovery_min_apply_delay', 'restore_command') and param not in wanted_recovery_params:
+                continue
             if param == 'recovery_min_apply_delay':
                 if not compare_values('integer', 'ms', value[0], wanted_recovery_params.get(param, 0)):
                     record_missmatch(value[1])
@@ -837,7 +869,7 @@ class ConfigHandler(object):
         self._postgresql.set_connection_kwargs(self.local_connect_kwargs)
 
     def _get_pg_settings(self, names):
-        return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context '
+        return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context, sourcefile'
                                                          + ' FROM pg_catalog.pg_settings ' +
                                                          ' WHERE pg_catalog.lower(name) = ANY(%s)'),
                                                         [n.lower() for n in names])}
@@ -866,9 +898,9 @@ class ConfigHandler(object):
         conf_changed = hba_changed = ident_changed = local_connection_address_changed = pending_restart = False
         if self._postgresql.state == 'running':
             changes = CaseInsensitiveDict({p: v for p, v in server_parameters.items()
-                                           if '.' not in p and p.lower() not in self._RECOVERY_PARAMETERS})
+                                           if p.lower() not in self._RECOVERY_PARAMETERS})
             changes.update({p: None for p in self._server_parameters.keys()
-                            if not ('.' in p or p in changes or p.lower() in self._RECOVERY_PARAMETERS)})
+                            if not (p in changes or p.lower() in self._RECOVERY_PARAMETERS)})
             if changes:
                 if 'wal_buffers' in changes:  # we need to calculate the default value of wal_buffers
                     undef = [p for p in ('shared_buffers', 'wal_segment_size', 'wal_block_size') if p not in changes]
@@ -894,21 +926,17 @@ class ConfigHandler(object):
                                     local_connection_address_changed = True
                             else:
                                 logger.info('Changed %s from %s to %s', r[0], r[1], new_value)
-                for param in changes:
-                    if param in server_parameters:
+                for param, value in changes.items():
+                    if '.' in param:
+                        # Check that user-defined-paramters have changed (parameters with period in name)
+                        if value is None or param not in self._server_parameters \
+                                or str(value) != str(self._server_parameters[param]):
+                            logger.info('Changed %s from %s to %s', param, self._server_parameters.get(param), value)
+                            conf_changed = True
+                    elif 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)
-            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])):
-                    logger.info('Changed %s from %s to %s', p, self._server_parameters.get(p), v)
-                    conf_changed = True
-            for p, v in self._server_parameters.items():
-                if '.' in p and (p not in server_parameters or str(v) != str(server_parameters[p])):
-                    logger.info('Changed %s from %s to %s', p, v, server_parameters.get(p))
-                    conf_changed = True
-
             if not server_parameters.get('hba_file') and config.get('pg_hba'):
                 hba_changed = self._config.get('pg_hba', []) != config['pg_hba']
 

patroni/postgresql/postmaster.py

@@ -20,9 +20,9 @@ elif sys.version_info >= (3, 4):  # pragma: no cover
 logger = logging.getLogger(__name__)
 
 STOP_SIGNALS = {
-    'smart': signal.SIGTERM,
-    'fast': signal.SIGINT,
-    'immediate': signal.SIGQUIT if os.name != 'nt' else signal.SIGABRT,
+    'smart': 'TERM',
+    'fast': 'INT',
+    'immediate': 'QUIT',
 }
 
 
@@ -105,7 +105,7 @@ class PostmasterProcess(psutil.Process):
         except psutil.NoSuchProcess:
             return None
 
-    def signal_stop(self, mode):
+    def signal_stop(self, mode, pg_ctl='pg_ctl'):
         """Signal postmaster process to stop
 
         :returns None if signaled, True if process is already gone, False if error
@@ -113,8 +113,10 @@ class PostmasterProcess(psutil.Process):
         if self.is_single_user:
             logger.warning("Cannot stop server; single-user server is running (PID: {0})".format(self.pid))
             return False
+        if os.name != 'posix':
+            return self.pg_ctl_kill(mode, pg_ctl)
         try:
-            self.send_signal(STOP_SIGNALS[mode])
+            self.send_signal(getattr(signal, 'SIG' + STOP_SIGNALS[mode]))
         except psutil.NoSuchProcess:
             return True
         except psutil.AccessDenied as e:
@@ -123,6 +125,16 @@ class PostmasterProcess(psutil.Process):
 
         return None
 
+    def pg_ctl_kill(self, mode, pg_ctl):
+        try:
+            status = subprocess.call([pg_ctl, "kill", STOP_SIGNALS[mode], str(self.pid)])
+        except OSError:
+            return False
+        if status == 0:
+            return None
+        else:
+            return not self.is_running()
+
     def wait_for_user_backends_to_close(self):
         # These regexps are cross checked against versions PostgreSQL 9.1 .. 11
         aux_proc_re = re.compile("(?:postgres:)( .*:)? (?:(?:archiver|startup|autovacuum launcher|autovacuum worker|"

patroni/postgresql/rewind.py

@@ -132,13 +132,9 @@ class Rewind(object):
         return leader and leader.conn_url and self._state == REWIND_STATUS.NEED
 
     def check_for_checkpoint_after_promote(self):
-        if self._state == REWIND_STATUS.INITIAL and self._postgresql.is_leader():
-            try:
-                timeline = int(self._postgresql.controldata().get("Latest checkpoint's TimeLineID"))
-                if self._postgresql.get_master_timeline() == timeline:
-                    self._state = REWIND_STATUS.CHECKPOINT
-            except (TypeError, ValueError):
-                logger.exception('Failed to parse timeline from pg_controldata output')
+        if self._state == REWIND_STATUS.INITIAL and self._postgresql.is_leader() and \
+                self._postgresql.get_master_timeline() == self._postgresql.pg_control_timeline():
+            self._state = REWIND_STATUS.CHECKPOINT
 
     def checkpoint_after_promote(self):
         return self._state == REWIND_STATUS.CHECKPOINT

patroni/utils.py

@@ -1,7 +1,9 @@
 import logging
+import os
 import platform
 import random
 import re
+import tempfile
 import time
 
 from dateutil import tz
@@ -416,3 +418,27 @@ def cluster_as_json(cluster):
         if cluster.failover.candidate:
             ret['scheduled_switchover']['to'] = cluster.failover.candidate
     return ret
+
+
+def is_subpath(d1, d2):
+    real_d1 = os.path.realpath(d1) + os.path.sep
+    real_d2 = os.path.realpath(os.path.join(real_d1, d2))
+    return os.path.commonprefix([real_d1, real_d2 + os.path.sep]) == real_d1
+
+
+def validate_directory(d, msg="{} {}"):
+    if not os.path.exists(d):
+        try:
+            os.makedirs(d)
+        except OSError as e:
+            logger.error(e)
+            raise PatroniException(msg.format(d, "couldn't create the directory"))
+    elif os.path.isdir(d):
+        try:
+            fd, tmpfile = tempfile.mkstemp(dir=d)
+            os.close(fd)
+            os.remove(tmpfile)
+        except OSError:
+            raise PatroniException(msg.format(d, "the directory is not writable"))
+    else:
+        raise PatroniException(msg.format(d, "is not a directory"))

patroni/version.py

@@ -1 +1 @@
-__version__ = '1.6.2'
+__version__ = '1.6.4'

tests/__init__.py

@@ -86,7 +86,7 @@ class MockCursor(object):
         elif sql.startswith('SELECT slot_name'):
             self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b')]
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
-            self.results = [(1, 2)]
+            self.results = [(1, 2, 1)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
             self.results = [(False, 2)]
         elif sql.startswith('SELECT pg_catalog.to_char'):
@@ -164,7 +164,8 @@ class PostgresInit(unittest.TestCase):
                    'search_path': 'public', 'hot_standby': 'on', 'max_wal_senders': 5,
                    'wal_keep_segments': 8, 'wal_log_hints': 'on', 'max_locks_per_transaction': 64,
                    'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 0,
-                   'track_commit_timestamp': 'off', 'unix_socket_directories': '/tmp', 'trigger_file': 'bla'}
+                   'track_commit_timestamp': 'off', 'unix_socket_directories': '/tmp', 'trigger_file': 'bla',
+                   'stats_temp_directory': '/tmp'}
 
     @patch('psycopg2.connect', psycopg2_connect)
     @patch.object(ConfigHandler, 'write_postgresql_conf', Mock())

tests/test_bootstrap.py

@@ -209,13 +209,13 @@ class TestBootstrap(BaseTestPostgresql):
         mock_cancellable_subprocess_call.assert_called()
         args, kwargs = mock_cancellable_subprocess_call.call_args
         self.assertTrue('PGPASSFILE' in kwargs['env'])
-        self.assertEqual(args[0], ['/bin/false', 'postgres://127.0.0.2:5432/postgres'])
+        self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=127.0.0.2 port=5432'])
 
         mock_cancellable_subprocess_call.reset_mock()
         self.p.config._local_address.pop('host')
         self.assertTrue(self.b.call_post_bootstrap({'post_init': '/bin/false'}))
         mock_cancellable_subprocess_call.assert_called()
-        self.assertEqual(mock_cancellable_subprocess_call.call_args[0][0], ['/bin/false', 'postgres://:5432/postgres'])
+        self.assertEqual(mock_cancellable_subprocess_call.call_args[0][0], ['/bin/false', 'dbname=postgres port=5432'])
 
         mock_cancellable_subprocess_call.side_effect = OSError
         self.assertFalse(self.b.call_post_bootstrap({'post_init': '/bin/false'}))

tests/test_ctl.py

@@ -590,3 +590,14 @@ class TestCtl(unittest.TestCase):
         mock_get_dcs.return_value.get_cluster = get_cluster_not_initialized_without_leader
         result = self.runner.invoke(ctl, ['reinit', 'dummy'])
         assert "cluster doesn\'t have any members" in result.output
+
+    @patch('time.sleep', Mock())
+    @patch('patroni.ctl.get_dcs')
+    def test_reinit_wait(self, mock_get_dcs):
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
+        with patch.object(PoolManager, 'request') as mocked:
+            mocked.side_effect = [Mock(data=s, status=200) for s in
+                                  [b"reinitialize", b'{"state":"creating replica"}', b'{"state":"running"}']]
+            result = self.runner.invoke(ctl, ['reinit', 'alpha', 'other', '--wait'], input='y\ny')
+        self.assertIn("Waiting for reinitialize to complete on: other", result.output)
+        self.assertIn("Reinitialize is completed on: other", result.output)

tests/test_ha.py

@@ -150,7 +150,7 @@ def run_async(self, func, args=()):
 
 @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
 @patch.object(Postgresql, 'is_leader', Mock(return_value=True))
-@patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10)))
+@patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1)))
 @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=3))
 @patch.object(Postgresql, 'call_nowait', Mock(return_value=True))
 @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
@@ -199,9 +199,12 @@ class TestHa(PostgresInit):
         self.assertTrue(self.ha.update_lock(True))
 
     def test_touch_member(self):
-        self.p.timeline_wal_position = Mock(return_value=(0, 1))
+        self.p.timeline_wal_position = Mock(return_value=(0, 1, 0))
         self.p.replica_cached_timeline = Mock(side_effect=Exception)
         self.ha.touch_member()
+        self.p.timeline_wal_position = Mock(return_value=(0, 1, 1))
+        self.p.set_role('standby_leader')
+        self.ha.touch_member()
 
     def test_is_leader(self):
         self.assertFalse(self.ha.is_leader())
@@ -601,7 +604,7 @@ class TestHa(PostgresInit):
         # in synchronous_mode consider itself healthy if the former leader is accessible in read-only and ahead of us
         with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)):
             self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        with patch('patroni.postgresql.Postgresql.timeline_wal_position', return_value=(1, 1)):
+        with patch('patroni.postgresql.Postgresql.timeline_wal_position', return_value=(1, 1, 1)):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         with patch('patroni.postgresql.Postgresql.replica_cached_timeline', return_value=1):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))

tests/test_kubernetes.py

@@ -46,7 +46,8 @@ class TestKubernetes(unittest.TestCase):
     @patch('kubernetes.client.api_client.ThreadPool', Mock(), create=True)
     @patch.object(Thread, 'start', Mock())
     def setUp(self):
-        self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, 'labels': {'f': 'b'}})
+        self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0',
+                             'loop_wait': 10, 'retry_timeout': 10, 'labels': {'f': 'b'}})
         self.assertRaises(AttributeError, self.k._pods._build_cache)
         self.k._pods._is_ready = True
         self.assertRaises(AttributeError, self.k._kinds._build_cache)
@@ -71,14 +72,14 @@ class TestKubernetes(unittest.TestCase):
     @patch('kubernetes.config.load_kube_config', Mock())
     @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', Mock())
     def test_update_leader(self):
-        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10,
+        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
                         'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
         self.assertIsNotNone(k.update_leader('123'))
 
     @patch('kubernetes.config.load_kube_config', Mock())
     @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', Mock())
     def test_update_leader_with_restricted_access(self):
-        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10,
+        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
                         'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
         self.assertIsNotNone(k.update_leader('123', True))
 
@@ -120,7 +121,7 @@ class TestKubernetes(unittest.TestCase):
     @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints',
                   Mock(side_effect=[k8s_client.rest.ApiException(502, ''), k8s_client.rest.ApiException(500, '')]))
     def test_delete_sync_state(self):
-        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10,
+        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
                         'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
         self.assertFalse(k.delete_sync_state())
 
@@ -139,7 +140,7 @@ class TestKubernetes(unittest.TestCase):
     @patch.object(k8s_client.CoreV1Api, 'create_namespaced_service',
                   Mock(side_effect=[True, False, k8s_client.rest.ApiException(500, '')]))
     def test__create_config_service(self):
-        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10,
+        k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
                         'labels': {'f': 'b'}, 'use_endpoints': True, 'pod_ip': '10.0.0.0'})
         self.assertIsNotNone(k.patch_or_create_config({'foo': 'bar'}))
         self.assertIsNotNone(k.patch_or_create_config({'foo': 'bar'}))
@@ -152,7 +153,8 @@ class TestCacheBuilder(unittest.TestCase):
     @patch('kubernetes.client.api_client.ThreadPool', Mock(), create=True)
     @patch.object(Thread, 'start', Mock())
     def setUp(self):
-        self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0', 'retry_timeout': 10, 'labels': {'f': 'b'}})
+        self.k = Kubernetes({'ttl': 30, 'scope': 'test', 'name': 'p-0',
+                             'loop_wait': 10, 'retry_timeout': 10, 'labels': {'f': 'b'}})
 
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map)
     @patch('patroni.dcs.kubernetes.ObjectCache._watch')

tests/test_patroni.py

@@ -174,6 +174,7 @@ class TestPatroni(unittest.TestCase):
     @patch.object(Thread, 'join', Mock())
     def test_shutdown(self):
         self.p.api.shutdown = Mock(side_effect=Exception)
+        self.p.ha.shutdown = Mock(side_effect=Exception)
         self.p.shutdown()
 
     def test_check_psycopg2(self):

tests/test_postgresql.py

@@ -8,7 +8,7 @@ import time
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 from patroni.async_executor import CriticalTask
 from patroni.dcs import Cluster, ClusterConfig, Member, RemoteMember, SyncState
-from patroni.exceptions import PostgresConnectionException
+from patroni.exceptions import PostgresConnectionException, PatroniException
 from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.postgresql.slots import SlotsHandler
@@ -206,14 +206,16 @@ class TestPostgresql(BaseTestPostgresql):
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test_check_recovery_conf(self, mock_get_pg_settings):
         mock_get_pg_settings.return_value = {
-            'primary_conninfo': ['primary_conninfo', 'foo=', None, 'string', 'postmaster'],
-            'recovery_min_apply_delay': ['recovery_min_apply_delay', '0', 'ms', 'integer', 'sighup']
+            'primary_conninfo': ['primary_conninfo', 'foo=', None, 'string', 'postmaster', self.p.config._auto_conf],
+            'recovery_min_apply_delay': ['recovery_min_apply_delay', '0', 'ms', 'integer', 'sighup', 'foo']
         }
         self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
         self.p.config.write_recovery_conf({'standby_mode': 'on'})
         self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
         mock_get_pg_settings.return_value['primary_conninfo'][1] = ''
         mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '1'
+        self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))
+        mock_get_pg_settings.return_value['recovery_min_apply_delay'][5] = self.p.config._auto_conf
         self.assertEqual(self.p.config.check_recovery_conf(None), (True, False))
         mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '0'
         self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))
@@ -234,7 +236,8 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(MockPostmaster, 'create_time', Mock(return_value=1234567), create=True)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test__read_recovery_params(self, mock_get_pg_settings):
-        mock_get_pg_settings.return_value = {'primary_conninfo': ['primary_conninfo', '', None, 'string', 'postmaster']}
+        mock_get_pg_settings.return_value = {'primary_conninfo': ['primary_conninfo', '', None, 'string',
+                                                                  'postmaster', self.p.config._postgresql_conf]}
         self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
         self.p.config.write_postgresql_conf()
         self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))
@@ -331,7 +334,7 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertTrue(self.p.promote(0))
 
     def test_timeline_wal_position(self):
-        self.assertEqual(self.p.timeline_wal_position(), (1, 2))
+        self.assertEqual(self.p.timeline_wal_position(), (1, 2, 1))
         Thread(target=self.p.timeline_wal_position).start()
 
     @patch.object(PostmasterProcess, 'from_pidfile')
@@ -695,3 +698,8 @@ class TestPostgresql(BaseTestPostgresql):
             self.p.cancellable.cancel()
             self.assertFalse(self.p.start())
             self.assertTrue(self.p.pending_restart)
+
+    @patch('os.path.exists', Mock(return_value=True))
+    @patch('os.path.isfile', Mock(return_value=False))
+    def test_pgpass_is_dir(self):
+        self.assertRaises(PatroniException, self.setUp)

tests/test_postmaster.py

@@ -66,6 +66,8 @@ class TestPostmasterProcess(unittest.TestCase):
     @patch('psutil.Process.__init__', Mock())
     @patch('psutil.Process.send_signal')
     @patch('psutil.Process.pid', Mock(return_value=123))
+    @patch('os.name', 'posix')
+    @patch('signal.SIGQUIT', 3, create=True)
     def test_signal_stop(self, mock_send_signal):
         proc = PostmasterProcess(-123)
         self.assertEqual(proc.signal_stop('immediate'), False)
@@ -76,6 +78,21 @@ class TestPostmasterProcess(unittest.TestCase):
         self.assertEqual(proc.signal_stop('immediate'), True)
         self.assertEqual(proc.signal_stop('immediate'), False)
 
+    @patch('psutil.Process.__init__', Mock())
+    @patch('patroni.postgresql.postmaster.os')
+    @patch('subprocess.call', Mock(side_effect=[0, OSError, 1]))
+    @patch('psutil.Process.pid', Mock(return_value=123))
+    @patch('psutil.Process.is_running', Mock(return_value=False))
+    def test_signal_stop_nt(self, mock_os):
+        mock_os.configure_mock(name="nt")
+        proc = PostmasterProcess(-123)
+        self.assertEqual(proc.signal_stop('immediate'), False)
+
+        proc = PostmasterProcess(123)
+        self.assertEqual(proc.signal_stop('immediate'), None)
+        self.assertEqual(proc.signal_stop('immediate'), False)
+        self.assertEqual(proc.signal_stop('immediate'), True)
+
     @patch('psutil.Process.__init__', Mock())
     @patch('psutil.wait_procs')
     def test_wait_for_user_backends_to_close(self, mock_wait):

tests/test_utils.py

@@ -2,7 +2,7 @@ import unittest
 
 from mock import Mock, patch
 from patroni.exceptions import PatroniException
-from patroni.utils import Retry, RetryFailedError, polling_loop
+from patroni.utils import Retry, RetryFailedError, polling_loop, validate_directory
 
 
 class TestUtils(unittest.TestCase):
@@ -10,6 +10,29 @@ class TestUtils(unittest.TestCase):
     def test_polling_loop(self):
         self.assertEqual(list(polling_loop(0.001, interval=0.001)), [0])
 
+    @patch('os.path.exists', Mock(return_value=True))
+    @patch('os.path.isdir', Mock(return_value=True))
+    @patch('tempfile.mkstemp', Mock(return_value=("", "")))
+    @patch('os.remove', Mock(side_effect=Exception))
+    def test_validate_directory_writable(self):
+        self.assertRaises(Exception, validate_directory, "/tmp")
+
+    @patch('os.path.exists', Mock(return_value=True))
+    @patch('os.path.isdir', Mock(return_value=True))
+    @patch('tempfile.mkstemp', Mock(side_effect=OSError))
+    def test_validate_directory_not_writable(self):
+        self.assertRaises(PatroniException, validate_directory, "/tmp")
+
+    @patch('os.path.exists', Mock(return_value=False))
+    @patch('os.makedirs', Mock(side_effect=OSError))
+    def test_validate_directory_couldnt_create(self):
+        self.assertRaises(PatroniException, validate_directory, "/tmp")
+
+    @patch('os.path.exists', Mock(return_value=True))
+    @patch('os.path.isdir', Mock(return_value=False))
+    def test_validate_directory_is_not_a_directory(self):
+        self.assertRaises(PatroniException, validate_directory, "/tmp")
+
 
 @patch('time.sleep', Mock())
 class TestRetrySleeper(unittest.TestCase):