Release 1.6.4 (#1380)

* Bump version
* Update release notes

docs/ENVIRONMENT.rst

@@ -15,6 +15,7 @@ Global/Universal
 Log
 ---
 -  **PATRONI\_LOG\_LEVEL**: sets the general logging level. Default value is **INFO** (see `the docs for Python logging <https://docs.python.org/3.6/library/logging.html#levels>`_)
+-  **PATRONI\_LOG\_TRACEBACK\_LEVEL**: sets the level where tracebacks will be visible. Default value is **ERROR**. Set it to **DEBUG** if you want to see tracebacks only if you enable **PATRONI\_LOG\_LEVEL=DEBUG**.
 -  **PATRONI\_LOG\_FORMAT**: sets the log formatting string. Default value is **%(asctime)s %(levelname)s: %(message)s** (see `the LogRecord attributes <https://docs.python.org/3.6/library/logging.html#logrecord-attributes>`_)
 -  **PATRONI\_LOG\_DATEFORMAT**: sets the datetime formatting string. (see the `formatTime() documentation <https://docs.python.org/3.6/library/logging.html#logging.Formatter.formatTime>`_)
 -  **PATRONI\_LOG\_MAX\_QUEUE\_SIZE**: Patroni is using two-step logging. Log records are written into the in-memory queue and there is a separate thread which pulls them from the queue and writes to stderr or file. The maximum size of the internal queue is limited by default by **1000** records, which is enough to keep logs for the past 1h20m.
@@ -45,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

@@ -46,6 +46,7 @@ Global/Universal
 Log
 ---
 -  **level**: sets the general logging level. Default value is **INFO** (see `the docs for Python logging <https://docs.python.org/3.6/library/logging.html#levels>`_)
+-  **traceback\_level**: sets the level where tracebacks will be visible. Default value is **ERROR**. Set it to **DEBUG** if you want to see tracebacks only if you enable **log.level=DEBUG**.
 -  **format**: sets the log formatting string. Default value is **%(asctime)s %(levelname)s: %(message)s** (see `the LogRecord attributes <https://docs.python.org/3.6/library/logging.html#logrecord-attributes>`_)
 -  **dateformat**: sets the datetime formatting string. (see the `formatTime() documentation <https://docs.python.org/3.6/library/logging.html#logging.Formatter.formatTime>`_)
 -  **max\_queue\_size**: Patroni is using two-step logging. Log records are written into the in-memory queue and there is a separate thread which pulls them from the queue and writes to stderr or file. The maximum size of the internal queue is limited by default by **1000** records, which is enough to keep logs for the past 1h20m.
@@ -97,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,7 +3,139 @@
 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
+-------------
+
+**New features**
+
+- Implemented ``patroni --version`` (Igor Yanchenko)
+
+  It prints the current version of Patroni and exits.
+
+- Set the ``user-agent`` http header for all http requests (Alexander Kukushkin)
+
+  Patroni is communicating with Consul, Etcd, and Kubernetes API via the http protocol. Having a specifically crafted ``user-agent`` (example: ``Patroni/1.6.2 Python/3.6.8 Linux``) might be useful for debugging and monitoring.
+
+- Make it possible to configure log level for exception tracebacks (Igor)
+
+  If you set ``log.traceback_level=DEBUG`` the tracebacks will be visible only when ``log.level=DEBUG``. The default behavior remains the same.
+
+
+**Stability improvements**
+
+- Avoid importing all DCS modules when searching for the module required by the config file (Alexander)
+
+  There is no need to import modules for Etcd, Consul, and Kubernetes if we need only e.g. Zookeeper. It helps to reduce memory usage and solves the problem of having INFO messages ``Failed to import smth``.
+
+- Removed python ``requests`` module from explicit requirements (Alexander)
+
+  It wasn't used for anything critical, but causing a lot of problems when the new version of ``urllib3`` is released.
+
+- Improve handling of ``etcd.hosts`` written as a comma-separated string instead of YAML array (Igor)
+
+  Previously it was failing when written in format ``host1:port1, host2:port2`` (the space character after the comma).
+
+
+**Usability improvements**
+
+- Don't force users to choose members from an empty list in ``patronictl`` (Igor)
+
+  If the user provides a wrong cluster name, we will raise an exception rather than ask to choose a member from an empty list.
+
+- Make the error message more helpful if the REST API cannot bind (Igor)
+
+  For an inexperienced user it might be hard to figure out what is wrong from the Python stacktrace.
+
+
+**Bugfixes**
+
+- Fix calculation of ``wal_buffers`` (Alexander)
+
+  The base unit has been changed from 8 kB blocks to bytes in PostgreSQL 11.
+
+- Use ``passfile`` in ``primary_conninfo`` only on PostgreSQL 10+ (Alexander)
+
+  On older versions there is no guarantee that ``passfile`` will work, unless the latest version of ``libpq`` is installed.
+
+
 Version 1.6.1
+-------------
 
 **New features**
 

docs/rest_api.rst

@@ -333,4 +333,4 @@ Reinitialize endpoint
 
 The call might fail if Patroni is in a loop trying to recover (restart) a failed Postgres. In order to overcome this problem one can specify ``{"force":true}`` in the request body.
 
-The reinitialize endpoint is used by ``patronictl reinitialize``.
+The reinitialize endpoint is used by ``patronictl reinit``.

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/basic_replication.feature

@@ -4,6 +4,7 @@ Feature: basic replication
   Scenario: check replication of a single table
     Given I start postgres0
     Then postgres0 is a leader after 10 seconds
+    And there is a non empty initialize key in DCS after 15 seconds
     When I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "loop_wait": 2, "synchronous_mode": true}
     Then I receive a response code 200
     When I start postgres1
@@ -35,10 +36,12 @@ Feature: basic replication
     And I run patronictl.py resume batman
     Then I receive a response returncode 0
     And postgres2 role is the primary after 24 seconds
+    And Response on GET http://127.0.0.1:8010/history contains recovery after 10 seconds
     When I issue a PATCH request to http://127.0.0.1:8010/config with {"synchronous_mode": null, "master_start_timeout": 0}
     Then I receive a response code 200
     When I add the table bar to postgres2
     Then table bar is present on postgres1 after 20 seconds
+    And Response on GET http://127.0.0.1:8010/config contains master_start_timeout after 10 seconds
 
   Scenario: check immediate failover when master_start_timeout=0
     Given I kill postmaster on postgres2

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/custom_bootstrap.feature

@@ -13,5 +13,5 @@ Scenario: make a backup and do a restore into a new cluster
     Given I add the table bar to postgres1
     And I do a backup of postgres1
     When I start postgres2 in a cluster batman2 from backup
-    Then postgres2 is a leader of batman2 after 10 seconds
+    Then postgres2 is a leader of batman2 after 30 seconds
     And table bar is present on postgres2 after 10 seconds

features/environment.py

@@ -1,11 +1,6 @@
 import abc
-import consul
 import datetime
-import etcd
-import kazoo.client
-import kazoo.exceptions
 import os
-import psutil
 import psycopg2
 import json
 import shutil
@@ -176,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',
@@ -249,59 +245,24 @@ class PatroniController(AbstractController):
         except Exception:
             return None
 
-    def database_is_running(self):
-        pid = self._get_pid()
-        if not pid:
-            return False
-        try:
-            os.kill(pid, 0)
-        except OSError:
-            return False
-        return True
-
     def patroni_hang(self, timeout):
         hang = ProcessHang(self._handle.pid, timeout)
         self._closables.append(hang)
         hang.start()
 
-    def checkpoint_hang(self, timeout):
-        pid = self._get_pid()
-        if not pid:
-            return False
-        proc = psutil.Process(pid)
-        for child in proc.children():
-            if 'checkpoint' in child.cmdline()[0]:
-                checkpointer = child
-                break
-        else:
-            return False
-        hang = ProcessHang(checkpointer.pid, timeout)
-        self._closables.append(hang)
-        hang.start()
-        return True
-
     def cancel_background(self):
         for obj in self._closables:
             obj.close()
         self._closables = []
 
-    def terminate_backends(self):
-        pid = self._get_pid()
-        if not pid:
-            return False
-        proc = psutil.Process(pid)
-        for p in proc.children():
-            if 'process' not in p.cmdline()[0]:
-                p.terminate()
-
     @property
     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):
@@ -375,9 +336,11 @@ class ConsulController(AbstractDcsController):
         super(ConsulController, self).__init__(context)
         os.environ['PATRONI_CONSUL_HOST'] = 'localhost:8500'
         os.environ['PATRONI_CONSUL_REGISTER_SERVICE'] = 'on'
-        self._client = consul.Consul()
         self._config_file = None
 
+        import consul
+        self._client = consul.Consul()
+
     def _start(self):
         self._config_file = self._work_directory + '.json'
         with open(self._config_file, 'wb') as f:
@@ -417,6 +380,8 @@ class EtcdController(AbstractDcsController):
     def __init__(self, context):
         super(EtcdController, self).__init__(context)
         os.environ['PATRONI_ETCD_HOST'] = 'localhost:2379'
+
+        import etcd
         self._client = etcd.Client(port=2379)
 
     def _start(self):
@@ -424,12 +389,14 @@ class EtcdController(AbstractDcsController):
                                 stdout=self._log, stderr=subprocess.STDOUT)
 
     def query(self, key, scope='batman'):
+        import etcd
         try:
             return self._client.get(self.path(key, scope)).value
         except etcd.EtcdKeyNotFound:
             return None
 
     def cleanup_service_tree(self):
+        import etcd
         try:
             self._client.delete(self.path(scope=''), recursive=True)
         except (etcd.EtcdKeyNotFound, etcd.EtcdConnectionFailed):
@@ -474,12 +441,12 @@ class KubernetesController(AbstractDcsController):
     def delete_pod(self, name):
         try:
             self._api.delete_namespaced_pod(name, self._namespace, body=self._client.V1DeleteOptions())
-        except:
+        except Exception:
             pass
         while True:
             try:
                 self._api.read_namespaced_pod(name, self._namespace)
-            except:
+            except Exception:
                 break
 
     def query(self, key, scope='batman'):
@@ -488,22 +455,23 @@ class KubernetesController(AbstractDcsController):
             return (pod.metadata.annotations or {}).get('status', '')
         else:
             try:
-                e = self._api.read_namespaced_endpoints(scope + ('' if key == 'leader' else '-' + key), self._namespace)
-                if key == 'leader':
+                ep = scope + {'leader': '', 'history': '-config', 'initialize': '-config'}.get(key, '-' + key)
+                e = self._api.read_namespaced_endpoints(ep, self._namespace)
+                if key != 'sync':
                     return e.metadata.annotations[key]
                 else:
                     return json.dumps(e.metadata.annotations)
-            except:
+            except Exception:
                 return None
 
     def cleanup_service_tree(self):
         try:
             self._api.delete_collection_namespaced_pod(self._namespace, label_selector=self._label_selector)
-        except:
+        except Exception:
             pass
         try:
             self._api.delete_collection_namespaced_endpoints(self._namespace, label_selector=self._label_selector)
-        except:
+        except Exception:
             pass
 
         while True:
@@ -523,18 +491,22 @@ class ZooKeeperController(AbstractDcsController):
         super(ZooKeeperController, self).__init__(context, False)
         if export_env:
             os.environ['PATRONI_ZOOKEEPER_HOSTS'] = "'localhost:2181'"
+
+        import kazoo.client
         self._client = kazoo.client.KazooClient()
 
     def _start(self):
         pass  # TODO: implement later
 
     def query(self, key, scope='batman'):
+        import kazoo.exceptions
         try:
             return self._client.get(self.path(key, scope))[0].decode('utf-8')
         except kazoo.exceptions.NoNodeError:
             return None
 
     def cleanup_service_tree(self):
+        import kazoo.exceptions
         try:
             self._client.delete(self.path(scope=''), recursive=True)
         except (kazoo.exceptions.NoNodeError):
@@ -561,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
@@ -594,9 +567,8 @@ class PatroniPoolController(object):
         self._processes[name].start(max_wait_limit)
 
     def __getattr__(self, func):
-        if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to', 'add_tag_to_config',
-                        'get_watchdog', 'database_is_running', 'checkpoint_hang', 'patroni_hang',
-                        'terminate_backends', 'backup']:
+        if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to',
+                        'add_tag_to_config', 'get_watchdog', 'patroni_hang', 'backup']:
             raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func))
 
         def wrapper(name, *args, **kwargs):
@@ -610,7 +582,7 @@ class PatroniPoolController(object):
         self._processes.clear()
 
     def create_and_set_output_directory(self, feature_name):
-        feature_dir = os.path.join(self.patroni_path, 'features/output', feature_name.replace(' ', '_'))
+        feature_dir = os.path.join(self.patroni_path, 'features', 'output', feature_name.replace(' ', '_'))
         if os.path.exists(feature_dir):
             shutil.rmtree(feature_dir)
         os.makedirs(feature_dir)
@@ -623,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': {
@@ -636,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'},
@@ -653,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/patroni_api.feature

@@ -43,6 +43,20 @@ Scenario: check dynamic configuration change via DCS
 	When I issue a GET request to http://127.0.0.1:8008/patroni
 	Then I receive a response code 200
 	And I receive a response tags {'new_tag': 'new_value'}
+	And I sleep for 4 seconds
+
+Scenario: check the scheduled restart
+	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"superuser_reserved_connections": "6"}}}
+	Then I receive a response code 200
+		And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 5 seconds
+	Given I issue a scheduled restart at http://127.0.0.1:8008 in 3 seconds with {"role": "replica"}
+	Then I receive a response code 202
+		And I sleep for 4 seconds
+		And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 10 seconds
+	Given I issue a scheduled restart at http://127.0.0.1:8008 in 3 seconds with {"restart_pending": "True"}
+	Then I receive a response code 202
+		And Response on GET http://127.0.0.1:8008/patroni does not contain pending_restart after 10 seconds
+		And postgres0 role is the primary after 10 seconds
 
 Scenario: check API requests for the primary-replica pair in the pause mode
 	Given I run patronictl.py pause batman
@@ -101,16 +115,3 @@ Scenario: check the scheduled switchover
 	Then I receive a response code 503
 	When I issue a GET request to http://127.0.0.1:8009/replica
 	Then I receive a response code 200
-
-Scenario: check the scheduled restart
-	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"superuser_reserved_connections": "6"}}}
-	Then I receive a response code 200
-		And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 5 seconds
-	Given I issue a scheduled restart at http://127.0.0.1:8008 in 3 seconds with {"role": "replica"}
-	Then I receive a response code 202
-		And I sleep for 4 seconds
-		And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 10 seconds
-	Given I issue a scheduled restart at http://127.0.0.1:8008 in 3 seconds with {"restart_pending": "True"}
-	Then I receive a response code 202
-		And Response on GET http://127.0.0.1:8008/patroni does not contain pending_restart after 10 seconds
-

features/standby_cluster.feature

@@ -2,7 +2,7 @@ Feature: standby cluster
   Scenario: check permanent logical slots are preserved on failover/switchover
     Given I start postgres1
     Then postgres1 is a leader after 10 seconds
-    And I sleep for 3 seconds
+    And there is a non empty initialize key in DCS after 15 seconds
     When I issue a PATCH request to http://127.0.0.1:8009/config with {"loop_wait": 2, "slots": {"pm_1": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}}
     Then I receive a response code 200
     And Response on GET http://127.0.0.1:8009/config contains slots after 10 seconds
@@ -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')
@@ -34,3 +34,17 @@ def check_member(context, name, key, value, time_limit):
             pass
         time.sleep(1)
     assert False, "{0} does not have {1}={2} in dcs after {3} seconds".format(name, key, value, time_limit)
+
+
+@step('there is a non empty {key:w} key in DCS after {time_limit:d} seconds')
+def check_initialize(context, key, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+    while time.time() < max_time:
+        try:
+            if context.dcs_ctl.query(key):
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, "There is no {0} in dcs after {1} seconds".format(key, time_limit)

features/steps/patroni_api.py

@@ -1,8 +1,6 @@
-import base64
 import json
 import os
 import parse
-import requests
 import shlex
 import subprocess
 import sys
@@ -12,8 +10,10 @@ import yaml
 from behave import register_type, step, then
 from dateutil import tz
 from datetime import datetime, timedelta
+from patroni.request import PatroniRequest
 
 tzutc = tz.tzutc()
+request_executor = PatroniRequest({'ctl': {'auth': 'username:password'}})
 
 
 @parse.with_pattern(r'https?://(?:\w|\.|:|/)+')
@@ -45,9 +45,9 @@ def sleep_for_n_seconds(context, value):
 
 
 def _set_response(context, response):
-    context.status_code = response.status_code
-    data = response.content.decode('utf-8')
-    ct = response.headers.get('content-type', '')
+    context.status_code = response.status
+    data = response.data.decode('utf-8')
+    ct = response.getheader('content-type', '')
     if ct.startswith('application/json') or\
             ct.startswith('text/yaml') or\
             ct.startswith('text/x-yaml') or\
@@ -63,13 +63,7 @@ def _set_response(context, response):
 
 @step('I issue a GET request to {url:url}')
 def do_get(context, url):
-    try:
-        r = requests.get(url)
-    except requests.exceptions.RequestException:
-        context.status_code = None
-        context.response = None
-    else:
-        _set_response(context, r)
+    do_request(context, 'GET', url, None)
 
 
 @step('I issue an empty POST request to {url:url}')
@@ -79,17 +73,11 @@ def do_post_empty(context, url):
 
 @step('I issue a {request_method:w} request to {url:url} with {data}')
 def do_request(context, request_method, url, data):
-    data = data and json.loads(data) or {}
-    headers = {'Authorization': 'Basic ' + base64.b64encode('username:password'.encode('utf-8')).decode('utf-8'),
-               'Content-Type': 'application/json'}
+    data = data and json.loads(data)
     try:
-        if request_method == 'PATCH':
-            r = requests.patch(url, headers=headers, json=data)
-        else:
-            r = requests.post(url, headers=headers, json=data)
-    except requests.exceptions.RequestException:
-        context.status_code = None
-        context.response = None
+        r = request_executor.request(request_method, url, data)
+    except Exception:
+        context.status_code = context.response = None
     else:
         _set_response(context, r)
 
@@ -149,8 +137,8 @@ def add_tag_to_config(context, tag, value, pg_name):
 def check_http_response(context, url, value, timeout, negate=False):
     timeout *= context.timeout_multiplier
     for _ in range(int(timeout)):
-        r = requests.get(url)
-        if (value in r.content.decode('utf-8')) != negate:
+        r = request_executor.request('GET', url)
+        if (value in r.data.decode('utf-8')) != negate:
             break
         time.sleep(1)
     else:

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

features/steps/watchdog.py

@@ -44,31 +44,6 @@ def watchdog_was_triggered(context, name, timeout):
     assert False
 
 
-@then('{name:w} watchdog was not triggered')
-def watchdog_was_not_triggered(context, name):
-    assert not context.pctl.get_watchdog(name).was_triggered
-
-
-@step('{name:w} checkpoint takes {timeout:d} seconds')
-def checkpoint_hang(context, name, timeout):
-    assert context.pctl.checkpoint_hang(name, timeout)
-
-
 @step('{name:w} hangs for {timeout:d} seconds')
 def patroni_hang(context, name, timeout):
     return context.pctl.patroni_hang(name, timeout)
-
-
-@step('I terminate {name:w} user processes')
-def terminate_backends(context, name):
-    return context.pctl.terminate_backends(name)
-
-
-@step('Sleep for {timeout:d} seconds')
-def dcs_connection_lost(context, timeout):
-    time.sleep(timeout)
-
-
-@then('{name:w} database is running')
-def database_is_running(context, name):
-    assert context.pctl.database_is_running(name)

patroni/__init__.py

@@ -4,27 +4,29 @@ import signal
 import sys
 import time
 
+from patroni.version import __version__
+
 logger = logging.getLogger(__name__)
 
+PATRONI_ENV_PREFIX = 'PATRONI_'
+
 
 class Patroni(object):
 
-    def __init__(self):
+    def __init__(self, conf):
         from patroni.api import RestApiServer
-        from patroni.config import Config
         from patroni.dcs import get_dcs
         from patroni.ha import Ha
         from patroni.log import PatroniLogger
         from patroni.postgresql import Postgresql
         from patroni.request import PatroniRequest
-        from patroni.version import __version__
         from patroni.watchdog import Watchdog
 
         self.setup_signal_handlers()
 
         self.version = __version__
         self.logger = PatroniLogger()
-        self.config = Config()
+        self.config = conf
         self.logger.reload_config(self.config.get('log', {}))
         self.dcs = get_dcs(self.config)
         self.watchdog = Watchdog(self.config)
@@ -157,12 +159,31 @@ 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()
 
 
 def patroni_main():
-    patroni = Patroni()
+    import argparse
+    from patroni.config import Config, ConfigParseError
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--version', action='version', version='%(prog)s {0}'.format(__version__))
+    parser.add_argument('configfile', nargs='?', default='',
+                        help='Patroni may also read the configuration from the {0} environment variable'
+                        .format(Config.PATRONI_CONFIG_VARIABLE))
+    args = parser.parse_args()
+    try:
+        conf = Config(args.configfile)
+    except ConfigParseError as e:
+        if e.value:
+            print(e.value)
+        parser.print_help()
+        sys.exit(1)
+    patroni = Patroni(conf)
     try:
         patroni.run()
     except KeyboardInterrupt:
@@ -176,13 +197,6 @@ def fatal(string, *args):
     sys.exit(1)
 
 
-def use_spawn_start_method():
-    if sys.version_info >= (3, 4):
-        # The default, forking, method is not a good idea in a multithreaded process: https://bugs.python.org/issue6721
-        import multiprocessing
-        multiprocessing.set_start_method('spawn')
-
-
 def check_psycopg2():
     min_psycopg2 = (2, 5, 4)
     min_psycopg2_str = '.'.join(map(str, min_psycopg2))
@@ -205,9 +219,8 @@ def check_psycopg2():
 
 
 def main():
-    use_spawn_start_method()
-    check_psycopg2()
     if os.getpid() != 1:
+        check_psycopg2()
         return patroni_main()
 
     # Patroni started with PID=1, it looks like we are in the container

patroni/api.py

@@ -10,8 +10,8 @@ import os
 import six
 import socket
 
-from patroni.postgresql import PostgresConnectionException
-from patroni.postgresql.misc import postgres_version_to_int, PostgresException
+from patroni.exceptions import PostgresConnectionException, PostgresException
+from patroni.postgresql.misc import postgres_version_to_int
 from patroni.utils import deep_compare, parse_bool, patch_config, Retry, \
     RetryFailedError, parse_int, split_host_port, tzutc, uri, cluster_as_json
 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
@@ -556,7 +556,12 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
         info.sort(key=lambda x: x[0] == socket.AF_INET, reverse=not dual_stack)
 
         self.address_family = info[0][0]
-        HTTPServer.__init__(self, info[0][-1][:2], RestApiHandler)
+        try:
+            HTTPServer.__init__(self, info[0][-1][:2], RestApiHandler)
+        except socket.error:
+            logger.error(
+                    "Couldn't start a service on '%s:%s', please check your `restapi.listen` configuration", host, port)
+            raise
 
     def __initialize(self, listen, ssl_options):
         try:

patroni/config.py

@@ -2,16 +2,16 @@ import json
 import logging
 import os
 import shutil
-import sys
 import tempfile
 import yaml
 
 from collections import defaultdict
 from copy import deepcopy
+from patroni import PATRONI_ENV_PREFIX
+from patroni.exceptions import ConfigParseError
 from patroni.dcs import ClusterConfig
-from patroni.postgresql.config import ConfigHandler
+from patroni.postgresql.config import CaseInsensitiveDict, ConfigHandler
 from patroni.utils import deep_compare, parse_bool, parse_int, patch_config
-from requests.structures import CaseInsensitiveDict
 
 logger = logging.getLogger(__name__)
 
@@ -26,6 +26,11 @@ _AUTH_ALLOWED_PARAMETERS = (
 )
 
 
+def default_validator(conf):
+    if not conf:
+        return "Config is empty."
+
+
 class Config(object):
     """
     This class is responsible for:
@@ -46,7 +51,6 @@ class Config(object):
          to work with it as with the old `config` object.
     """
 
-    PATRONI_ENV_PREFIX = 'PATRONI_'
     PATRONI_CONFIG_VARIABLE = PATRONI_ENV_PREFIX + 'CONFIGURATION'
 
     __CACHE_FILENAME = 'patroni.dynamic.json'
@@ -76,27 +80,26 @@ class Config(object):
         }
     }
 
-    def __init__(self):
+    def __init__(self, configfile, validator=default_validator):
         self._modify_index = -1
         self._dynamic_configuration = {}
 
         self.__environment_configuration = self._build_environment_configuration()
 
         # Patroni reads the configuration from the command-line argument if it exists, otherwise from the environment
-        self._config_file = len(sys.argv) >= 2 and os.path.isfile(sys.argv[1]) and sys.argv[1]
+        self._config_file = configfile and os.path.isfile(configfile) and configfile
         if self._config_file:
             self._local_configuration = self._load_config_file()
         else:
             config_env = os.environ.pop(self.PATRONI_CONFIG_VARIABLE, None)
             self._local_configuration = config_env and yaml.safe_load(config_env) or self.__environment_configuration
-            if not self._local_configuration:
-                print('Usage: {0} config.yml'.format(sys.argv[0]))
-                print('\tPatroni may also read the configuration from the {0} environment variable'.
-                      format(self.PATRONI_CONFIG_VARIABLE))
-                sys.exit(1)
+        if validator:
+            error = validator(self._local_configuration)
+            if error:
+                raise ConfigParseError(error)
 
         self.__effective_configuration = self._build_effective_configuration({}, self._local_configuration)
-        self._data_dir = self.__effective_configuration['postgresql']['data_dir']
+        self._data_dir = self.__effective_configuration.get('postgresql', {}).get('data_dir', "")
         self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME)
         self._load_cache()
         self._cache_needs_saving = False
@@ -214,7 +217,7 @@ class Config(object):
         ret = defaultdict(dict)
 
         def _popenv(name):
-            return os.environ.pop(Config.PATRONI_ENV_PREFIX + name.upper(), None)
+            return os.environ.pop(PATRONI_ENV_PREFIX + name.upper(), None)
 
         for param in ('name', 'namespace', 'scope'):
             value = _popenv(param)
@@ -223,7 +226,7 @@ class Config(object):
 
         def _fix_log_env(name, oldname):
             value = _popenv(oldname)
-            name = Config.PATRONI_ENV_PREFIX + 'LOG_' + name.upper()
+            name = PATRONI_ENV_PREFIX + 'LOG_' + name.upper()
             if value and name not in os.environ:
                 os.environ[name] = value
 
@@ -239,7 +242,7 @@ class Config(object):
         _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile', 'cafile', 'verify_client'])
         _set_section_values('ctl', ['insecure', 'cacert', 'certfile', 'keyfile'])
         _set_section_values('postgresql', ['listen', 'connect_address', 'config_dir', 'data_dir', 'pgpass', 'bin_dir'])
-        _set_section_values('log', ['level', 'format', 'dateformat', 'max_queue_size',
+        _set_section_values('log', ['level', 'traceback_level', 'format', 'dateformat', 'max_queue_size',
                                     'dir', 'file_size', 'file_num', 'loggers'])
 
         def _parse_dict(value):
@@ -288,7 +291,7 @@ class Config(object):
                 return None
 
         for param in list(os.environ.keys()):
-            if param.startswith(Config.PATRONI_ENV_PREFIX):
+            if param.startswith(PATRONI_ENV_PREFIX):
                 # PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT|..)
                 name, suffix = (param[8:].split('_', 1) + [''])[:2]
                 if suffix in ('HOST', 'HOSTS', 'PORT', 'USE_PROXIES', 'PROTOCOL', 'SRV', 'URL', 'PROXY',
@@ -311,7 +314,7 @@ class Config(object):
 
         users = {}
         for param in list(os.environ.keys()):
-            if param.startswith(Config.PATRONI_ENV_PREFIX):
+            if param.startswith(PATRONI_ENV_PREFIX):
                 name, suffix = (param[8:].rsplit('_', 1) + [''])[:2]
                 # PATRONI_<username>_PASSWORD=<password>, PATRONI_<username>_OPTIONS=<option1,option2,...>
                 # CREATE USER "<username>" WITH <OPTIONS> PASSWORD '<password>'
@@ -341,7 +344,7 @@ class Config(object):
                 config[name] = deepcopy(value) if value else {}
 
         # restapi server expects to get restapi.auth = 'username:password'
-        if 'authentication' in config['restapi']:
+        if 'restapi' in config and 'authentication' in config['restapi']:
             config['restapi']['auth'] = '{username}:{password}'.format(**config['restapi']['authentication'])
 
         # special treatment for old config

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 == '':
@@ -72,17 +72,7 @@ def load_config(path, dcs):
         logging.debug('Ignoring configuration file "%s". It does not exists or is not readable.', path)
     else:
         logging.debug('Loading configuration from file %s', path)
-    config = {}
-    old_argv = list(sys.argv)
-    try:
-        sys.argv[1] = path
-        if Config.PATRONI_CONFIG_VARIABLE not in os.environ:
-            for p in ('PATRONI_RESTAPI_LISTEN', 'PATRONI_POSTGRESQL_DATA_DIR'):
-                if p not in os.environ:
-                    os.environ[p] = '.'
-        config = Config().copy()
-    finally:
-        sys.argv = old_argv
+    config = Config(path, validator=None).copy()
 
     dcs = parse_dcs(dcs) or parse_dcs(config.get('dcs_api')) or {}
     if dcs:
@@ -247,10 +237,12 @@ def get_cursor(cluster, connect_parameters, role='master', member=None):
     return None
 
 
-def get_members(cluster, cluster_name, member_names, role, force, action, scheduled_at=None):
+def get_members(cluster, cluster_name, member_names, role, force, action, ask_confirmation=True):
     candidates = {m.name: m for m in cluster.members}
 
     if not force or role:
+        if not member_names and not candidates:
+            raise PatroniCtlException('{0} cluster doesn\'t have any members'.format(cluster_name))
         output_members(cluster, cluster_name)
 
     if role:
@@ -270,21 +262,26 @@ def get_members(cluster, cluster_name, member_names, role, force, action, schedu
         if member_name not in candidates:
             raise PatroniCtlException('{0} is not a member of cluster'.format(member_name))
 
+    members = [candidates[n] for n in member_names]
+    if ask_confirmation:
+        confirm_members_action(members, force, action)
+    return members
+
+
+def confirm_members_action(members, force, action, scheduled_at=None):
     if scheduled_at:
         if not force:
             confirm = click.confirm('Are you sure you want to schedule {0} of members {1} at {2}?'
-                                    .format(action, ', '.join(member_names), scheduled_at))
+                                    .format(action, ', '.join([m.name for m in members]), scheduled_at))
             if not confirm:
                 raise PatroniCtlException('Aborted scheduled {0}'.format(action))
     else:
         if not force:
             confirm = click.confirm('Are you sure you want to {0} members {1}?'
-                                    .format(action, ', '.join(member_names)))
+                                    .format(action, ', '.join([m.name for m in members])))
             if not confirm:
                 raise PatroniCtlException('Aborted {0}'.format(action))
 
-    return [candidates[n] for n in member_names]
-
 
 @ctl.command('dsn', help='Generate a dsn for the provided member, defaults to a dsn of the master')
 @click.option('--role', '-r', help='Give a dsn of any member with this role', type=click.Choice(['master', 'replica',
@@ -501,14 +498,15 @@ def reload(obj, cluster_name, member_names, force, role):
 def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, version, pending, timeout):
     cluster = get_dcs(obj, cluster_name).get_cluster()
 
+    members = get_members(cluster, cluster_name, member_names, role, force, 'restart', False)
     if scheduled is None and not force:
         next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M')
         scheduled = click.prompt('When should the restart take place (e.g. ' + next_hour + ') ',
                                  type=str, default='now')
 
     scheduled_at = parse_scheduled(scheduled)
+    confirm_members_action(members, force, 'restart', scheduled_at)
 
-    members = get_members(cluster, cluster_name, member_names, role, force, 'restart', scheduled_at)
     if p_any:
         random.shuffle(members)
         members = members[:1]
@@ -560,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

@@ -73,26 +73,34 @@ def dcs_modules():
 
 
 def get_dcs(config):
-    available_implementations = set()
-    for module_name in dcs_modules():
-        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
+    modules = dcs_modules()
+
+    for module_name in modules:
+        name = module_name.split('.')[-1]
+        if name in config:  # we will try to import only modules which have configuration section in the config file
+            try:
+                module = importlib.import_module(module_name)
+                for key, item in module.__dict__.items():  # iterate through the module content
+                    # try to find implementation of AbstractDCS interface, class name must match with module_name
+                    if key.lower() == name and inspect.isclass(item) and issubclass(item, AbstractDCS):
                         # 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:
+                logger.debug('Failed to import %s', module_name)
+
+    available_implementations = []
+    for module_name in modules:
+        name = module_name.split('.')[-1]
+        try:
+            module = importlib.import_module(module_name)
+            available_implementations.extend(name for key, item in module.__dict__.items() if key.lower() == name
+                                             and inspect.isclass(item) and issubclass(item, AbstractDCS))
         except ImportError:
-            if not config.get('patronictl'):
-                logger.info('Failed to import %s', module_name)
+            logger.info('Failed to import %s', module_name)
     raise PatroniException("""Can not find suitable configuration of distributed configuration store
-Available implementations: """ + ', '.join(available_implementations))
+Available implementations: """ + ', '.join(sorted(set(available_implementations))))
 
 
 class Member(namedtuple('Member', 'index,name,session,data')):
@@ -159,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
@@ -232,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

@@ -9,13 +9,14 @@ import time
 import urllib3
 
 from consul import ConsulException, NotFound, base
-from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
-from patroni.exceptions import DCSError
-from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri
 from urllib3.exceptions import HTTPError
 from six.moves.urllib.parse import urlencode, urlparse, quote
 from six.moves.http_client import HTTPException
 
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, TimelineHistory
+from ..exceptions import DCSError
+from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT
+
 logger = logging.getLogger(__name__)
 
 
@@ -111,8 +112,9 @@ class HTTPClient(object):
             else:
                 kwargs['timeout'] = self._read_timeout
             token = params.pop('token', self.token) if isinstance(params, dict) else self.token
+            kwargs['headers'] = urllib3.make_headers(user_agent=USER_AGENT)
             if token:
-                kwargs['headers'] = {'X-Consul-Token': token}
+                kwargs['headers']['X-Consul-Token'] = token
             return callback(self.response(self.http.request(method.upper(), self.uri(path, params), **kwargs)))
         return wrapper
 
@@ -215,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

@@ -5,23 +5,23 @@ import logging
 import os
 import urllib3.util.connection
 import random
-import requests
 import six
 import socket
 import time
 
 from dns.exception import DNSException
 from dns import resolver
-from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
-from patroni.exceptions import DCSError
-from patroni.utils import Retry, RetryFailedError, split_host_port, uri
 from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError
-from requests.exceptions import RequestException
 from six.moves.queue import Queue
 from six.moves.http_client import HTTPException
 from six.moves.urllib_parse import urlparse
 from threading import Thread
 
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, TimelineHistory
+from ..exceptions import DCSError
+from ..request import get as requests_get
+from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT
+
 logger = logging.getLogger(__name__)
 
 
@@ -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 []
 
 
@@ -126,6 +126,10 @@ class Client(etcd.Client):
 
         return etcd_nodes, per_node_timeout, per_node_retries - 1
 
+    def _get_headers(self):
+        basic_auth = ':'.join((self.username, self.password)) if self.username and self.password else None
+        return urllib3.make_headers(basic_auth=basic_auth, user_agent=USER_AGENT)
+
     def _build_request_parameters(self, timeout=None):
         kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect}
 
@@ -284,12 +288,12 @@ class Client(etcd.Client):
                 url = uri(protocol, (host, port), endpoint)
                 if endpoint:
                     try:
-                        response = requests.get(url, timeout=self.read_timeout, verify=False)
-                        if response.ok:
-                            for member in response.json():
+                        response = requests_get(url, timeout=self.read_timeout, verify=False)
+                        if response.status < 400:
+                            for member in json.loads(response.data.decode('utf-8')):
                                 ret.extend(member['clientURLs'])
                             break
-                    except RequestException:
+                    except Exception:
                         logger.exception('GET %s', url)
                 else:
                     ret.append(url)
@@ -416,7 +420,7 @@ class Etcd(AbstractDCS):
             config['hosts'] = []
             for value in hosts:
                 if isinstance(value, six.string_types):
-                    config['hosts'].append(uri(protocol, split_host_port(value, default_port)))
+                    config['hosts'].append(uri(protocol, split_host_port(value.strip(), default_port)))
         elif 'host' in config:
             host, port = split_host_port(config['host'], 2379)
             config['host'] = host

patroni/dcs/exhibitor.py

@@ -1,11 +1,11 @@
+import json
 import logging
 import random
-import requests
 import time
 
 from patroni.dcs.zookeeper import ZooKeeper
+from patroni.request import get as requests_get
 from patroni.utils import uri
-from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)
 
@@ -48,10 +48,10 @@ class ExhibitorEnsembleProvider(object):
         random.shuffle(exhibitors)
         for host in exhibitors:
             try:
-                response = requests.get(uri('http', (host, self._exhibitor_port), self._uri_path), timeout=self.TIMEOUT)
-                return response.json()
-            except RequestException:
-                pass
+                response = requests_get(uri('http', (host, self._exhibitor_port), self._uri_path), timeout=self.TIMEOUT)
+                return json.loads(response.data.decode('utf-8'))
+            except Exception:
+                logging.debug('Request to %s failed', host)
         return None
 
     @property

patroni/dcs/kubernetes.py

@@ -8,14 +8,15 @@ import sys
 import time
 
 from kubernetes import client as k8s_client, config as k8s_config, watch as k8s_watch
-from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
-from patroni.exceptions import DCSError
-from patroni.utils import deep_compare, tzutc, Retry, RetryFailedError
 from urllib3 import Timeout
 from urllib3.exceptions import HTTPError
 from six.moves.http_client import HTTPException
 from threading import Condition, Lock, Thread
 
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, TimelineHistory
+from ..exceptions import DCSError
+from ..utils import deep_compare, Retry, RetryFailedError, tzutc, USER_AGENT
+
 logger = logging.getLogger(__name__)
 
 
@@ -35,12 +36,27 @@ class CoreV1ApiProxy(object):
 
     def __init__(self, use_endpoints=False):
         self._api = k8s_client.CoreV1Api()
+        self._api.api_client.user_agent = USER_AGENT
         self._api.api_client.rest_client.pool_manager.connection_pool_kw['maxsize'] = 10
         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'):
@@ -208,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
@@ -248,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):
@@ -481,7 +499,8 @@ class Kubernetes(AbstractDCS):
         """Unused"""
 
     def manual_failover(self, leader, candidate, scheduled_at=None, index=None):
-        annotations = {'leader': leader or None, 'member': candidate or None, 'scheduled_at': scheduled_at}
+        annotations = {'leader': leader or None, 'member': candidate or None,
+                       'scheduled_at': scheduled_at and scheduled_at.isoformat()}
         patch = bool(self.cluster and isinstance(self.cluster.failover, Failover) and self.cluster.failover.index)
         return self.patch_or_create(self.failover_path, annotations, index, bool(index or patch), False)
 

patroni/exceptions.py

@@ -27,3 +27,7 @@ class PostgresConnectionException(PostgresException):
 
 class WatchdogError(PatroniException):
     pass
+
+
+class ConfigParseError(PatroniException):
+    pass

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/log.py

@@ -11,6 +11,20 @@ from threading import Lock, Thread
 _LOGGER = logging.getLogger(__name__)
 
 
+def debug_exception(logger_obj, msg, *args, **kwargs):
+    kwargs.pop("exc_info", False)
+    if logger_obj.isEnabledFor(logging.DEBUG):
+        logger_obj.debug(msg, *args, exc_info=True, **kwargs)
+    else:
+        msg = "{0}, DETAIL: '{1}'".format(msg, sys.exc_info()[1])
+        logger_obj.error(msg, *args, exc_info=False, **kwargs)
+
+
+def error_exception(logger_obj, msg, *args, **kwargs):
+    exc_info = kwargs.pop("exc_info", True)
+    logger_obj.error(msg, *args, exc_info=exc_info, **kwargs)
+
+
 class QueueHandler(logging.Handler):
 
     def __init__(self):
@@ -61,6 +75,7 @@ class ProxyHandler(logging.Handler):
 class PatroniLogger(Thread):
 
     DEFAULT_LEVEL = 'INFO'
+    DEFAULT_TRACEBACK_LEVEL = 'ERROR'
     DEFAULT_FORMAT = '%(asctime)s %(levelname)s: %(message)s'
 
     NORMAL_LOG_QUEUE_SIZE = 2  # When everything goes normal Patroni writes only 2 messages per HA loop
@@ -99,6 +114,10 @@ class PatroniLogger(Thread):
                 self._queue_handler.queue.maxsize = config.get('max_queue_size', self.DEFAULT_MAX_QUEUE_SIZE)
 
             self._root_logger.setLevel(config.get('level', PatroniLogger.DEFAULT_LEVEL))
+            if config.get('traceback_level', PatroniLogger.DEFAULT_TRACEBACK_LEVEL).lower() == 'debug':
+                logging.Logger.exception = debug_exception
+            else:
+                logging.Logger.exception = error_exception
 
             new_handler = None
             if 'dir' in config:

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):
@@ -627,9 +636,8 @@ class Postgresql(object):
         # Don't try to call pg_controldata during backup restore
         if self._version_file_exists() and self.state != 'creating replica':
             try:
-                env = {'LANG': 'C', 'LC_ALL': 'C', 'PATH': os.getenv('PATH')}
-                if os.getenv('SYSTEMROOT') is not None:
-                    env['SYSTEMROOT'] = os.getenv('SYSTEMROOT')
+                env = os.environ.copy()
+                env.update(LANG='C', LC_ALL='C')
                 data = subprocess.check_output([self.pgcommand('pg_controldata'), self._data_dir], env=env)
                 if data:
                     data = data.decode('utf-8').splitlines()
@@ -734,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:
@@ -816,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 requests.structures import CaseInsensitiveDict
+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__)
 
@@ -250,6 +252,21 @@ class ConfigWriter(object):
         self.writeline("{0} = '{1}'".format(param, self.escape(value)))
 
 
+class CaseInsensitiveDict(HTTPHeaderDict):
+
+    def add(self, key, val):
+        self[key] = val
+
+    def __getitem__(self, key):
+        return self._container[key.lower()][1]
+
+    def __repr__(self):
+        return str(dict(self.items()))
+
+    def copy(self):
+        return CaseInsensitiveDict(self._container.values())
+
+
 class ConfigHandler(object):
 
     # List of parameters which must be always passed to postmaster as command line options
@@ -318,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
@@ -332,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)]
@@ -421,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'],
@@ -465,22 +500,27 @@ 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', 'host', 'port', 'sslmode', 'sslcompression', 'sslcert',
-                    'sslkey', 'sslrootcert', 'sslcrl', 'application_name', 'krbsrvname')
+        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()):
             if name == 'primary_conninfo':
-                if 'password' in value:
+                if 'password' in value and self._postgresql.major_version >= 100000:
                     self.write_pgpass(value)
                     value['passfile'] = self._passfile = self._pgpass
                     self._passfile_mtime = mtime(self._pgpass)
@@ -501,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):
@@ -544,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
@@ -653,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])
@@ -822,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])}
@@ -831,7 +878,8 @@ class ConfigHandler(object):
     def _handle_wal_buffers(old_values, changes):
         wal_block_size = parse_int(old_values['wal_block_size'][1])
         wal_segment_size = old_values['wal_segment_size']
-        wal_segment_size = parse_int(wal_segment_size[1]) * parse_int(wal_segment_size[2], 'B') / wal_block_size
+        wal_segment_unit = parse_int(wal_segment_size[2], 'B') if wal_segment_size[2][0].isdigit() else 1
+        wal_segment_size = parse_int(wal_segment_size[1]) * wal_segment_unit / wal_block_size
         default_wal_buffers = min(max(parse_int(old_values['shared_buffers'][1]) / 32, 8), wal_segment_size)
 
         wal_buffers = old_values['wal_buffers']
@@ -850,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]
@@ -878,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

@@ -5,15 +5,24 @@ import psutil
 import re
 import signal
 import subprocess
+import sys
 
-from patroni.config import Config
+from patroni import PATRONI_ENV_PREFIX
+
+# avoid spawning the resource tracker process
+if sys.version_info >= (3, 8):  # pragma: no cover
+    import multiprocessing.resource_tracker
+    multiprocessing.resource_tracker.getfd = lambda: 0
+elif sys.version_info >= (3, 4):  # pragma: no cover
+    import multiprocessing.semaphore_tracker
+    multiprocessing.semaphore_tracker.getfd = lambda: 0
 
 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',
 }
 
 
@@ -96,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
@@ -104,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:
@@ -114,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|"
@@ -155,7 +176,7 @@ class PostmasterProcess(psutil.Process):
         # In order to make everything portable we can't use fork&exec approach here, so  we will call
         # ourselves and pass list of arguments which must be used to start postgres.
         # On Windows, in order to run a side-by-side assembly the specified env must include a valid SYSTEMROOT.
-        env = {p: os.environ[p] for p in os.environ if not p.startswith(Config.PATRONI_ENV_PREFIX)}
+        env = {p: os.environ[p] for p in os.environ if not p.startswith(PATRONI_ENV_PREFIX)}
         try:
             proc = PostmasterProcess._from_pidfile(data_dir)
             if proc and not proc._is_postmaster_process():
@@ -172,8 +193,9 @@ class PostmasterProcess(psutil.Process):
             pass
         cmdline = [pgcommand, '-D', data_dir, '--config-file={}'.format(conf)] + options
         logger.debug("Starting postgres: %s", " ".join(cmdline))
-        parent_conn, child_conn = multiprocessing.Pipe(False)
-        proc = multiprocessing.Process(target=pg_ctl_start, args=(child_conn, cmdline, env))
+        ctx = multiprocessing.get_context('spawn') if sys.version_info >= (3, 4) else multiprocessing
+        parent_conn, child_conn = ctx.Pipe(False)
+        proc = ctx.Process(target=pg_ctl_start, args=(child_conn, cmdline, env))
         proc.start()
         pid = parent_conn.recv()
         proc.join()

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/request.py

@@ -1,14 +1,17 @@
 import json
 import urllib3
+import six
 
 from six.moves.urllib_parse import urlparse, urlunparse
 
+from .utils import USER_AGENT
+
 
 class PatroniRequest(object):
 
     def __init__(self, config, insecure=False):
         cert_reqs = 'CERT_NONE' if insecure or config.get('ctl', {}).get('insecure', False) else 'CERT_REQUIRED'
-        self._pool = urllib3.PoolManager(cert_reqs=cert_reqs)
+        self._pool = urllib3.PoolManager(num_pools=10, maxsize=10, cert_reqs=cert_reqs)
         self.reload_config(config)
 
     @staticmethod
@@ -27,7 +30,7 @@ class PatroniRequest(object):
         return value
 
     def reload_config(self, config):
-        self._pool.headers = urllib3.make_headers(basic_auth=self._get_cfg_value(config, 'auth'))
+        self._pool.headers = urllib3.make_headers(basic_auth=self._get_cfg_value(config, 'auth'), user_agent=USER_AGENT)
 
         if self._apply_ssl_file_param(config, 'cert'):
             self._apply_ssl_file_param(config, 'key')
@@ -37,11 +40,19 @@ class PatroniRequest(object):
         cacert = config.get('ctl', {}).get('cacert') or config.get('restapi', {}).get('cafile')
         self._apply_pool_param('ca_certs', cacert)
 
+    def request(self, method, url, body=None, **kwargs):
+        if body is not None and not isinstance(body, six.string_types):
+            body = json.dumps(body)
+        return self._pool.request(method.upper(), url, body=body, **kwargs)
+
     def __call__(self, member, method='GET', endpoint=None, data=None, **kwargs):
         url = member.api_url
         if endpoint:
             scheme, netloc, _, _, _, _ = urlparse(url)
             url = urlunparse((scheme, netloc, endpoint, '', '', ''))
-        if data is not None:
-            kwargs['body'] = json.dumps(data)
-        return self._pool.request(method.upper(), url, **kwargs)
+        return self.request(method, url, data, **kwargs)
+
+
+def get(url, verify=True, **kwargs):
+    http = PatroniRequest({}, not verify)
+    return http.request('GET', url, **kwargs)

patroni/scripts/aws.py

@@ -1,12 +1,12 @@
 #!/usr/bin/env python
 
+import json
 import logging
-import requests
-from requests.exceptions import RequestException
 import sys
 import boto.ec2
 
 from patroni.utils import Retry, RetryFailedError
+from patroni.request import get as requests_get
 
 logger = logging.getLogger(__name__)
 
@@ -19,14 +19,14 @@ class AWSConnection(object):
         self._retry = Retry(deadline=300, max_delay=30, max_tries=-1, retry_exceptions=(boto.exception.StandardError,))
         try:
             # get the instance id
-            r = requests.get('http://169.254.169.254/latest/dynamic/instance-identity/document', timeout=2.1)
-        except RequestException:
+            r = requests_get('http://169.254.169.254/latest/dynamic/instance-identity/document', timeout=2.1)
+        except Exception:
             logger.error('cannot query AWS meta-data')
             return
 
-        if r.ok:
+        if r.status < 400:
             try:
-                content = r.json()
+                content = json.loads(r.data.decode('utf-8'))
                 self.instance_id = content['instanceId']
                 self.region = content['region']
             except Exception:

patroni/utils.py

@@ -1,15 +1,21 @@
 import logging
+import os
+import platform
 import random
 import re
+import tempfile
 import time
 
 from dateutil import tz
-from patroni.exceptions import PatroniException
+
+from .exceptions import PatroniException
+from .version import __version__
 
 tzutc = tz.tzutc()
 
 logger = logging.getLogger(__name__)
 
+USER_AGENT = 'Patroni/{0} Python/{1} {2}'.format(__version__, platform.python_version(), platform.system())
 OCT_RE = re.compile(r'^[-+]?0[0-7]*')
 DEC_RE = re.compile(r'^[-+]?(0|[1-9][0-9]*)')
 HEX_RE = re.compile(r'^[-+]?0x[0-9a-fA-F]+')
@@ -412,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.1'
+__version__ = '1.6.4'

patroni/watchdog/base.py

@@ -57,7 +57,7 @@ class WatchdogConfig(object):
         return not self == other
 
     def get_impl(self):
-        if self.driver == 'testing':
+        if self.driver == 'testing':  # pragma: no cover
             from patroni.watchdog.linux import TestingWatchdogDevice
             return TestingWatchdogDevice.from_config(self.driver_config)
         elif platform.system() == 'Linux' and self.driver == 'default':

patroni/watchdog/linux.py

@@ -16,11 +16,11 @@ IOC_DIRBITS = 2
 
 # Non-generic platform special cases
 machine = platform.machine()
-if machine in ['mips', 'sparc', 'powerpc', 'ppc64']:
+if machine in ['mips', 'sparc', 'powerpc', 'ppc64']:  # pragma: no cover
     IOC_SIZEBITS = 13
     IOC_DIRBITS = 3
     IOC_NONE, IOC_WRITE, IOC_READ = 1, 2, 4
-elif machine == 'parisc':
+elif machine == 'parisc':  # pragma: no cover
     IOC_WRITE, IOC_READ = 2, 1
 
 IOC_NRSHIFT = 0
@@ -218,7 +218,7 @@ class LinuxWatchdogDevice(WatchdogBase):
         return timeout.value
 
 
-class TestingWatchdogDevice(LinuxWatchdogDevice):
+class TestingWatchdogDevice(LinuxWatchdogDevice):  # pragma: no cover
     """Converts timeout ioctls to regular writes that can be intercepted from a named pipe."""
     timeout = 60
 

postgres0.yml

@@ -17,7 +17,18 @@ restapi:
 #   cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 
 etcd:
+  #Provide host to do the initial discovery of the cluster topology:
   host: 127.0.0.1:2379
+  #Or use "hosts" to provide multiple endpoints
+  #Could be a comma separated string:
+  #hosts: host1:port1,host2:port2
+  #or an actual yaml list:
+  #hosts:
+  #- host1:port1
+  #- host2:port2
+  #Once discovery is complete Patroni will use the list of advertised clientURLs
+  #It is possible to change this behavior through by setting:
+  #use_proxies: true
 
 bootstrap:
   # this section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster

postgres1.yml

@@ -17,7 +17,18 @@ restapi:
 #   cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 
 etcd:
+  #Provide host to do the initial discovery of the cluster topology:
   host: 127.0.0.1:2379
+  #Or use "hosts" to provide multiple endpoints
+  #Could be a comma separated string:
+  #hosts: host1:port1,host2:port2
+  #or an actual yaml list:
+  #hosts:
+  #- host1:port1
+  #- host2:port2
+  #Once discovery is complete Patroni will use the list of advertised clientURLs
+  #It is possible to change this behavior through by setting:
+  #use_proxies: true
 
 bootstrap:
   # this section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster

postgres2.yml

@@ -17,7 +17,18 @@ restapi:
 #   cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 
 etcd:
+  #Provide host to do the initial discovery of the cluster topology:
   host: 127.0.0.1:2379
+  #Or use "hosts" to provide multiple endpoints
+  #Could be a comma separated string:
+  #hosts: host1:port1,host2:port2
+  #or an actual yaml list:
+  #hosts:
+  #- host1:port1
+  #- host2:port2
+  #Once discovery is complete Patroni will use the list of advertised clientURLs
+  #It is possible to change this behavior through by setting:
+  #use_proxies: true
 
 bootstrap:
   # this section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster

requirements.txt

@@ -1,7 +1,6 @@
 urllib3>=1.19.1,!=1.21
 boto
 PyYAML
-requests
 six >= 1.7
 kazoo>=1.3.1
 python-etcd>=0.4.3,<0.5

setup.py

@@ -8,22 +8,12 @@ import inspect
 import os
 import sys
 
-from patroni import check_psycopg2, fatal
-from patroni.version import __version__ as VERSION
-from setuptools.command.test import test as TestCommand
-from setuptools import find_packages, setup
-
-if sys.version_info < (2, 7, 0):
-    fatal('patroni needs to be run with Python 2.7+')
-check_psycopg2()
-del sys.modules['patroni']
-del sys.modules['patroni.version']
+from setuptools import Command, find_packages, setup
 
 __location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.currentframe())))
 
 NAME = 'patroni'
 MAIN_PACKAGE = NAME
-SCRIPTS = 'scripts'
 DESCRIPTION = 'PostgreSQL High-Available orchestrator and CLI'
 LICENSE = 'The MIT License'
 URL = 'https://github.com/zalando/patroni'
@@ -32,9 +22,10 @@ AUTHOR_EMAIL = '[email protected], [email protected], alexk
 KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
     ' zookeeper exhibitor consul streaming replication kubernetes k8s'
 
+EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'],
+                  'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['kubernetes']}
 COVERAGE_XML = True
 COVERAGE_HTML = False
-JUNIT_XML = True
 
 # Add here all kinds of additional classifiers as defined under
 # https://pypi.python.org/pypi?%3Aaction=list_classifiers
@@ -64,78 +55,74 @@ CONSOLE_SCRIPTS = ['patroni = patroni:main',
                    "patroni_aws = patroni.scripts.aws:main"]
 
 
-class PyTest(TestCommand):
+class PyTest(Command):
 
-    user_options = [('cov=', None, 'Run coverage'), ('cov-xml=', None, 'Generate junit xml report'), ('cov-html=',
-                    None, 'Generate junit html report'), ('junitxml=', None, 'Generate xml of test results')]
+    user_options = [('cov=', None, 'Run coverage'), ('cov-xml=', None, 'Generate junit xml report'),
+                    ('cov-html=', None, 'Generate junit html report')]
 
     def initialize_options(self):
-        TestCommand.initialize_options(self)
+        self.cov = []
         self.cov_xml = False
         self.cov_html = False
-        self.junitxml = None
 
     def finalize_options(self):
-        TestCommand.finalize_options(self)
         if self.cov_xml or self.cov_html:
             self.cov = ['--cov', MAIN_PACKAGE, '--cov-report', 'term-missing']
             if self.cov_xml:
                 self.cov.extend(['--cov-report', 'xml'])
             if self.cov_html:
                 self.cov.extend(['--cov-report', 'html'])
-        if self.junitxml is not None:
-            self.junitxml = ['--junitxml', self.junitxml]
 
     def run_tests(self):
         try:
             import pytest
         except Exception:
             raise RuntimeError('py.test is not installed, run: pip install pytest')
-        params = {'args': self.test_args}
-        if self.cov:
-            params['args'] += self.cov
-        if self.junitxml:
-            params['args'] += self.junitxml
-        params['args'] += ['--doctest-modules', MAIN_PACKAGE, '-vv']
 
         import logging
         silence = logging.WARNING
         logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=os.getenv('LOGLEVEL', silence))
-        params['args'] += ['-s' if logging.getLogger().getEffectiveLevel() < silence else '--capture=fd']
-        if not os.getenv('SYSTEMROOT'):
-            os.environ['SYSTEMROOT'] = '/'
-        errno = pytest.main(**params)
+
+        args = ['--verbose', 'tests', '--doctest-modules', MAIN_PACKAGE] +\
+            ['-s' if logging.getLogger().getEffectiveLevel() < silence else '--capture=fd']
+        if self.cov:
+            args += self.cov
+
+        errno = pytest.main(args=args)
         sys.exit(errno)
 
+    def run(self):
+        from pkg_resources import evaluate_marker
+        requirements = self.distribution.install_requires + ['mock>=2.0.0', 'pytest-cov', 'pytest'] +\
+            [v for k, v in self.distribution.extras_require.items() if not k.startswith(':') or evaluate_marker(k[1:])]
+        self.distribution.fetch_build_eggs(requirements)
+        self.run_tests()
+
 
 def read(fname):
     with open(os.path.join(__location__, fname)) as fd:
         return fd.read()
 
 
-def setup_package():
+def setup_package(version):
     # Assemble additional setup commands
     cmdclass = {'test': PyTest}
 
     install_requires = []
-    extras_require = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'],
-                      'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['kubernetes']}
 
     for r in read('requirements.txt').split('\n'):
         r = r.strip()
         if r == '':
             continue
         extra = False
-        for e, v in extras_require.items():
+        for e, v in EXTRAS_REQUIRE.items():
             if r.startswith(v[0]):
-                extras_require[e] = [r]
+                EXTRAS_REQUIRE[e] = [r]
                 extra = True
         if not extra:
             install_requires.append(r)
 
-    command_options = {'test': {'test_suite': ('setup.py', 'tests')}}
-    if JUNIT_XML:
-        command_options['test']['junitxml'] = 'setup.py', 'junit.xml'
+    command_options = {'test': {}}
     if COVERAGE_XML:
         command_options['test']['cov_xml'] = 'setup.py', True
     if COVERAGE_HTML:
@@ -143,7 +130,7 @@ def setup_package():
 
     setup(
         name=NAME,
-        version=VERSION,
+        version=version,
         url=URL,
         author=AUTHOR,
         author_email=AUTHOR_EMAIL,
@@ -152,17 +139,28 @@ def setup_package():
         keywords=KEYWORDS,
         long_description=read('README.rst'),
         classifiers=CLASSIFIERS,
-        test_suite='tests',
         packages=find_packages(exclude=['tests', 'tests.*']),
         package_data={MAIN_PACKAGE: ["*.json"]},
+        python_requires='>=2.7',
         install_requires=install_requires,
-        extras_require=extras_require,
+        extras_require=EXTRAS_REQUIRE,
+        setup_requires='flake8',
         cmdclass=cmdclass,
-        tests_require=['flake8', 'mock>=2.0.0', 'pytest-cov', 'pytest'],
         command_options=command_options,
         entry_points={'console_scripts': CONSOLE_SCRIPTS},
     )
 
 
 if __name__ == '__main__':
-    setup_package()
+    old_modules = sys.modules.copy()
+    try:
+        from patroni import check_psycopg2, fatal, __version__
+    finally:
+        sys.modules.clear()
+        sys.modules.update(old_modules)
+
+    if sys.version_info < (2, 7, 0):
+        fatal('Patroni needs to be run with Python 2.7+')
+    check_psycopg2()
+
+    setup_package(__version__)

tests/__init__.py

@@ -1,5 +1,4 @@
 import datetime
-import json
 import os
 import shutil
 import unittest
@@ -7,7 +6,7 @@ import unittest
 from mock import Mock, patch
 
 import psycopg2
-import requests
+import urllib3
 
 from patroni.dcs import Leader, Member
 from patroni.postgresql import Postgresql
@@ -24,19 +23,11 @@ class MockResponse(object):
     def __init__(self, status_code=200):
         self.status_code = status_code
         self.content = '{}'
-        self.ok = True
-
-    def json(self):
-        return json.loads(self.content)
 
     @property
     def data(self):
         return self.content.encode('utf-8')
 
-    @property
-    def text(self):
-        return self.content
-
     @property
     def status(self):
         return self.status_code
@@ -51,7 +42,7 @@ def requests_get(url, **kwargs):
               '"name":"default","clientURLs":["http://localhost:2379","http://localhost:4001"]}]'
     response = MockResponse()
     if url.startswith('http://local'):
-        raise requests.exceptions.RequestException()
+        raise urllib3.exceptions.HTTPError()
     elif ':8011/patroni' in url:
         response.content = '{"role": "replica", "xlog": {"received_location": 0}, "tags": {}}'
     elif url.endswith('/members'):
@@ -62,11 +53,9 @@ def requests_get(url, **kwargs):
         data = kwargs.get('data', '')
         if ' false}' in data:
             response.status_code = 503
-            response.ok = False
             response.content = 'restarting after failure already in progress'
     else:
         response.status_code = 404
-        response.ok = False
     return response
 
 
@@ -97,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'):
@@ -175,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())
@@ -183,7 +173,7 @@ class PostgresInit(unittest.TestCase):
     @patch.object(ConfigHandler, 'replace_pg_ident', Mock())
     @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='master'))
     def setUp(self):
-        data_dir = 'data/test0'
+        data_dir = os.path.join('data', 'test0')
         self.p = Postgresql({'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir,
                              'config_dir': data_dir, 'retry_timeout': 10,
                              'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'),

tests/test_api.py

@@ -438,3 +438,7 @@ class TestRestApiServer(unittest.TestCase):
             raise Exception()
         except Exception:
             self.assertIsNone(MockRestApiServer.handle_error(None, ('127.0.0.1', 55555)))
+
+    def test_socket_error(self):
+        with patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock(side_effect=socket.error)):
+            self.assertRaises(socket.error, MockRestApiServer, Mock(), '', {'listen': '*:8008'})

tests/test_aws.py

@@ -1,11 +1,11 @@
 import boto.ec2
 import sys
 import unittest
+import urllib3
 
 from mock import Mock, patch
 from collections import namedtuple
 from patroni.scripts.aws import AWSConnection, main as _main
-from requests.exceptions import RequestException
 
 
 class MockEc2Connection(object):
@@ -22,28 +22,11 @@ class MockEc2Connection(object):
         return True
 
 
-class MockResponse(object):
-    ok = True
-
-    def __init__(self, content):
-        self.content = content
-
-    def json(self):
-        return self.content
-
-
-def requests_get(url, **kwargs):
-    if url.split('/')[-1] == 'document':
-        result = {"instanceId": "012345", "region": "eu-west-1"}
-    else:
-        result = 'foo'
-    return MockResponse(result)
-
-
 @patch('boto.ec2.connect_to_region', Mock(return_value=MockEc2Connection()))
 class TestAWSConnection(unittest.TestCase):
 
-    @patch('requests.get', requests_get)
+    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(
+        status=200, body=b'{"instanceId": "012345", "region": "eu-west-1"}')))
     def setUp(self):
         self.conn = AWSConnection('test')
 
@@ -53,17 +36,18 @@ class TestAWSConnection(unittest.TestCase):
             self.conn._retry.max_tries = 1
             self.assertFalse(self.conn.on_role_change('master'))
 
-    @patch('requests.get', Mock(side_effect=RequestException('foo')))
+    @patch('patroni.scripts.aws.requests_get', Mock(side_effect=Exception('foo')))
     def test_non_aws(self):
         conn = AWSConnection('test')
         self.assertFalse(conn.on_role_change("master"))
 
-    @patch('requests.get', Mock(return_value=MockResponse('foo')))
+    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(status=200, body=b'foo')))
     def test_aws_bizare_response(self):
         conn = AWSConnection('test')
         self.assertFalse(conn.aws_available())
 
-    @patch('requests.get', requests_get)
+    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(
+        status=200, body=b'{"instanceId": "012345", "region": "eu-west-1"}')))
     @patch('sys.exit', Mock())
     def test_main(self):
         self.assertIsNone(_main())

tests/test_bootstrap.py

@@ -112,7 +112,8 @@ class TestBootstrap(BaseTestPostgresql):
         config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}}
 
         with patch.object(Postgresql, 'is_running', Mock(return_value=False)),\
-                patch('multiprocessing.Process', Mock(side_effect=Exception)):
+                patch('multiprocessing.Process', Mock(side_effect=Exception)),\
+                patch('multiprocessing.get_context', Mock(side_effect=Exception), create=True):
             self.assertRaises(Exception, self.b.bootstrap, config)
         with open(os.path.join(self.p.data_dir, 'pg_hba.conf')) as f:
             lines = f.readlines()
@@ -140,6 +141,7 @@ class TestBootstrap(BaseTestPostgresql):
 
         mock_cancellable_subprocess_call.return_value = 0
         with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))),\
+                patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True),\
                 patch('os.path.isfile', Mock(return_value=True)),\
                 patch('os.unlink', Mock()),\
                 patch.object(ConfigHandler, 'save_configuration_files', Mock()),\
@@ -207,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_config.py

@@ -15,10 +15,7 @@ class TestConfig(unittest.TestCase):
     def setUp(self):
         sys.argv = ['patroni.py']
         os.environ[Config.PATRONI_CONFIG_VARIABLE] = 'restapi: {}\npostgresql: {data_dir: foo}'
-        self.config = Config()
-
-    def test_no_config(self):
-        self.assertRaises(SystemExit, Config)
+        self.config = Config(None)
 
     def test_set_dynamic_configuration(self):
         with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
@@ -66,8 +63,7 @@ class TestConfig(unittest.TestCase):
             'PATRONI_admin_PASSWORD': 'admin',
             'PATRONI_admin_OPTIONS': 'createrole,createdb'
         })
-        sys.argv = ['patroni.py', 'postgres0.yml']
-        config = Config()
+        config = Config('postgres0.yml')
         with patch.object(Config, '_load_config_file', Mock(return_value={'restapi': {}})):
             with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
                 config.reload_local_configuration()

tests/test_ctl.py

@@ -1,6 +1,5 @@
 import etcd
 import os
-import sys
 import unittest
 
 from click.testing import CliRunner
@@ -25,7 +24,6 @@ CONFIG_FILE_PATH = './test-ctl.yaml'
 def test_rw_config():
     runner = CliRunner()
     with runner.isolated_filesystem():
-        sys.argv = ['patronictl.py', '']
         load_config(CONFIG_FILE_PATH + '/dummy', None)
         store_config({'etcd': {'host': 'localhost:2379'}}, CONFIG_FILE_PATH + '/dummy')
         load_config(CONFIG_FILE_PATH + '/dummy', '0.0.0.0')
@@ -159,6 +157,7 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['failover', 'dummy'], input='\n')
         assert 'Failover could be performed only to a specific candidate' in result.output
 
+    @patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd']))
     def test_get_dcs(self):
         self.assertRaises(PatroniCtlException, get_dcs, {'dummy': {}}, 'dummy')
 
@@ -584,3 +583,21 @@ class TestCtl(unittest.TestCase):
             self.assertIsNone(find_executable('vim'))
         with patch('os.path.isfile', Mock(side_effect=[False, True])):
             self.assertEqual(find_executable('vim', '/'), '/vim.exe')
+
+    @patch('patroni.ctl.get_dcs')
+    def test_get_members(self, mock_get_dcs):
+        mock_get_dcs.return_value = self.e
+        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_etcd.py

@@ -116,12 +116,12 @@ class TestDnsCachingResolver(unittest.TestCase):
 
 @patch('dns.resolver.query', dns_query)
 @patch('socket.getaddrinfo', socket_getaddrinfo)
-@patch('requests.get', requests_get)
+@patch('patroni.dcs.etcd.requests_get', requests_get)
 class TestClient(unittest.TestCase):
 
     @patch('dns.resolver.query', dns_query)
     @patch('socket.getaddrinfo', socket_getaddrinfo)
-    @patch('requests.get', requests_get)
+    @patch('patroni.dcs.etcd.requests_get', requests_get)
     def setUp(self):
         with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
@@ -199,7 +199,7 @@ class TestClient(unittest.TestCase):
                           socket_options=[(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)])
 
 
-@patch('requests.get', requests_get)
+@patch('patroni.dcs.etcd.requests_get', requests_get)
 @patch('socket.getaddrinfo', socket_getaddrinfo)
 @patch.object(etcd.Client, 'write', etcd_write)
 @patch.object(etcd.Client, 'read', etcd_read)

tests/test_exhibitor.py

@@ -1,4 +1,5 @@
 import unittest
+import urllib3
 
 from mock import Mock, patch
 from patroni.dcs.exhibitor import ExhibitorEnsembleProvider, Exhibitor
@@ -8,7 +9,7 @@ from . import SleepException, requests_get
 from .test_zookeeper import MockKazooClient
 
 
-@patch('requests.get', requests_get)
+@patch('patroni.dcs.exhibitor.requests_get', requests_get)
 @patch('time.sleep', Mock(side_effect=SleepException))
 class TestExhibitorEnsembleProvider(unittest.TestCase):
 
@@ -21,7 +22,8 @@ class TestExhibitorEnsembleProvider(unittest.TestCase):
 
 class TestExhibitor(unittest.TestCase):
 
-    @patch('requests.get', requests_get)
+    @patch('urllib3.PoolManager.request', Mock(return_value=urllib3.HTTPResponse(
+        status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}')))
     @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
     def setUp(self):
         self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test',

tests/test_ha.py

@@ -122,7 +122,7 @@ zookeeper:
         # all the extra values that are coming from py.test
         sys.argv = sys.argv[:1]
 
-        self.config = Config()
+        self.config = Config(None)
         self.config.set_dynamic_configuration({'maximum_lag_on_failover': 5})
         self.version = '1.5.7'
         self.postgresql = p
@@ -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))
@@ -177,7 +177,7 @@ def run_async(self, func, args=()):
 class TestHa(PostgresInit):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
-    @patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.foo', 'patroni.dcs.etcd']))
+    @patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.etcd']))
     @patch.object(etcd.Client, 'read', etcd_read)
     def setUp(self):
         super(TestHa, self).setUp()
@@ -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,22 +153,24 @@ 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')
     def test__build_cache(self, mock_response):
-        mock_response.return_value.read_chunked.return_value = json.dumps(
+        mock_response.return_value.read_chunked.return_value = [json.dumps(
             {'type': 'MODIFIED', 'object': {'metadata': {
                 'name': self.k.config_path, 'resourceVersion': '2', 'annotations': {self.k._CONFIG: 'foo'}}}}
-        ) + '\n' + json.dumps(
+        ).encode('utf-8'), ('\n' + json.dumps(
             {'type': 'DELETED', 'object': {'metadata': {
                 'name': self.k.config_path, 'resourceVersion': '3'}}}
         ) + '\n' + json.dumps(
             {'type': 'MDIFIED', 'object': {'metadata': {'name': self.k.config_path}}}
-        ) + '\n' + json.dumps({'object': {'code': 410}}) + '\n'
+        ) + '\n' + json.dumps({'object': {'code': 410}}) + '\n').encode('utf-8')]
         self.k._kinds._build_cache()
 
     @patch('patroni.dcs.kubernetes.logger.error', Mock(side_effect=SleepException))
+    @patch('patroni.dcs.kubernetes.ObjectCache._build_cache', Mock(side_effect=Exception))
     def test_run(self):
         self.assertRaises(SleepException, self.k._pods.run)

tests/test_log.py

@@ -9,6 +9,8 @@ from patroni.config import Config
 from patroni.log import PatroniLogger
 from six.moves.queue import Queue, Full
 
+_LOG = logging.getLogger(__name__)
+
 
 class TestPatroniLogger(unittest.TestCase):
 
@@ -22,6 +24,7 @@ class TestPatroniLogger(unittest.TestCase):
     def test_patroni_logger(self):
         config = {
             'log': {
+                'traceback_level': 'DEBUG',
                 'max_queue_size': 5,
                 'dir': 'foo',
                 'file_size': 4096,
@@ -35,8 +38,9 @@ class TestPatroniLogger(unittest.TestCase):
         sys.argv = ['patroni.py']
         os.environ[Config.PATRONI_CONFIG_VARIABLE] = yaml.dump(config, default_flow_style=False)
         logger = PatroniLogger()
-        patroni_config = Config()
+        patroni_config = Config(None)
         logger.reload_config(patroni_config['log'])
+        _LOG.exception('test')
         logger.start()
 
         with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)):
@@ -45,12 +49,13 @@ class TestPatroniLogger(unittest.TestCase):
         self.assertEqual(logger.log_handler.maxBytes, config['log']['file_size'])
         self.assertEqual(logger.log_handler.backupCount, config['log']['file_num'])
 
+        config['log']['level'] = 'DEBUG'
         config['log'].pop('dir')
         with patch('logging.Handler.close', Mock(side_effect=Exception)):
             logger.reload_config(config['log'])
             with patch.object(logging.Logger, 'makeRecord',
                               Mock(side_effect=[logging.LogRecord('', logging.INFO, '', 0, '', (), None), Exception])):
-                logging.error('test')
+                logging.exception('test')
             logging.error('test')
             with patch.object(Queue, 'put_nowait', Mock(side_effect=Full)):
                 self.assertRaises(SystemExit, logger.shutdown)

tests/test_patroni.py

@@ -2,10 +2,10 @@ import etcd
 import logging
 import os
 import signal
-import sys
 import time
 import unittest
 
+import patroni.config as config
 from mock import Mock, PropertyMock, patch
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
@@ -41,21 +41,23 @@ class MockFrozenImporter(object):
 @patch.object(etcd.Client, 'read', etcd_read)
 class TestPatroni(unittest.TestCase):
 
+    def test_no_config(self):
+        self.assertRaises(SystemExit, patroni_main)
+
     @patch('pkgutil.get_importer', Mock(return_value=MockFrozenImporter()))
     @patch('sys.frozen', Mock(return_value=True), create=True)
     @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
     @patch.object(etcd.Client, 'read', etcd_read)
     @patch.object(Thread, 'start', Mock())
+    @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
     def setUp(self):
         self._handlers = logging.getLogger().handlers[:]
         RestApiServer._BaseServer__is_shut_down = Mock()
         RestApiServer._BaseServer__shutdown_request = True
         RestApiServer.socket = 0
-        with patch.object(Client, 'machines') as mock_machines:
-            mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
-            sys.argv = ['patroni.py', 'postgres0.yml']
-            os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0'
-            self.p = Patroni()
+        os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0'
+        conf = config.Config('postgres0.yml')
+        self.p = Patroni(conf)
 
     def tearDown(self):
         logging.getLogger().handlers[:] = self._handlers
@@ -66,24 +68,23 @@ class TestPatroni(unittest.TestCase):
         self.p.load_dynamic_configuration()
         self.p.load_dynamic_configuration()
 
+    @patch('sys.argv', ['patroni.py', 'postgres0.yml'])
     @patch('time.sleep', Mock(side_effect=SleepException))
     @patch.object(etcd.Client, 'delete', Mock())
-    @patch.object(Client, 'machines')
+    @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
     @patch.object(Thread, 'join', Mock())
-    def test_patroni_patroni_main(self, mock_machines):
+    def test_patroni_patroni_main(self):
         with patch('subprocess.call', Mock(return_value=1)):
-            sys.argv = ['patroni.py', 'postgres0.yml']
-
-            mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             with patch.object(Patroni, 'run', Mock(side_effect=SleepException)):
+                os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0'
                 self.assertRaises(SleepException, patroni_main)
             with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
                 with patch('patroni.ha.Ha.is_paused', Mock(return_value=True)):
+                    os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0'
                     patroni_main()
 
     @patch('os.getpid')
     @patch('multiprocessing.Process')
-    @patch('patroni.use_spawn_start_method', Mock())
     @patch('patroni.patroni_main', Mock())
     def test_patroni_main(self, mock_process, mock_getpid):
         mock_getpid.return_value = 2
@@ -173,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

@@ -1,13 +1,14 @@
 import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
 import psycopg2
+import re
 import subprocess
 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
@@ -205,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))
@@ -220,7 +223,8 @@ class TestPostgresql(BaseTestPostgresql):
         with patch('patroni.postgresql.config.ConfigHandler.primary_conninfo_params', Mock(return_value=conninfo)):
             mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '1'
             self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
-            mock_get_pg_settings.return_value['primary_conninfo'][1] = 'host=1 passfile=' + self.p.config._pgpass
+            mock_get_pg_settings.return_value['primary_conninfo'][1] = 'host=1 passfile='\
+                + re.sub(r'([\'\\ ])', r'\\\1', self.p.config._pgpass)
             mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '0'
             self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
             self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': conninfo.copy()})
@@ -232,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))
@@ -329,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')
@@ -620,7 +625,7 @@ class TestPostgresql(BaseTestPostgresql):
         config['synchronous_mode_strict'] = True
         self.p.config.get_server_parameters(config)
         self.p.config.set_synchronous_standby('foo')
-        self.p.config.get_server_parameters(config)
+        self.assertTrue(str(self.p.config.get_server_parameters(config)).startswith('{'))
 
     @patch('time.sleep', Mock())
     def test__wait_for_connection_close(self):
@@ -693,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

@@ -1,3 +1,4 @@
+import multiprocessing
 import psutil
 import unittest
 
@@ -65,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)
@@ -75,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):
@@ -96,6 +114,7 @@ class TestPostmasterProcess(unittest.TestCase):
     @patch('subprocess.Popen')
     @patch('os.setsid', Mock(), create=True)
     @patch('multiprocessing.Process', MockProcess)
+    @patch('multiprocessing.get_context', Mock(return_value=multiprocessing), create=True)
     @patch.object(PostmasterProcess, 'from_pid')
     @patch.object(PostmasterProcess, '_from_pidfile')
     def test_start(self, mock_frompidfile, mock_frompid, mock_popen):

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