BUGFIX: don't leak password when running pg_rewind (#1321)

In addition to that:
* enforce security settings from `postgresql.authention`
* update release notes
* bump version
* close https://github.com/zalando/patroni/issues/1320

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.

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.

docs/releases.rst

@@ -3,7 +3,77 @@
 Release notes
 =============
 
+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/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/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
@@ -249,51 +244,16 @@ 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)
@@ -375,9 +335,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 +379,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 +388,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 +440,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 +454,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 +490,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):
@@ -594,9 +565,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 +580,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)
@@ -637,7 +607,7 @@ class PatroniPoolController(object):
                 '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'))
+                            os.path.join(self.patroni_path, 'data', 'wal_archive'))
                 },
                 'authentication': {
                     'superuser': {'password': 'zalando1'},
@@ -654,7 +624,7 @@ class PatroniPoolController(object):
                 'method': 'backup_restore',
                 'backup_restore': {
                     'command': 'features/backup_restore.sh --sourcedir=' + os.path.join(self.patroni_path,
-                                                                                        'data/basebackup'),
+                                                                                        'data', 'basebackup'),
                     'recovery_conf': {
                         'recovery_target_action': 'promote',
                         'recovery_target_timeline': 'latest',

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

features/steps/cascading_replication.py

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

@@ -31,7 +31,7 @@ def start_patroni(context, name, cluster_name):
             "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")}
+                                                                                    '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)
@@ -162,7 +164,23 @@ class Patroni(object):
 
 
 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 +194,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 +216,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

@@ -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]

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

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
 

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__)
 
 
@@ -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,6 +36,7 @@ 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
@@ -481,7 +483,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/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

@@ -627,9 +627,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()

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

patroni/postgresql/config.py

@@ -6,8 +6,8 @@ import socket
 import stat
 import time
 
-from requests.structures import CaseInsensitiveDict
 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
@@ -250,6 +250,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
@@ -465,22 +480,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)
@@ -831,7 +851,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']

patroni/postgresql/postmaster.py

@@ -5,8 +5,17 @@ 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__)
 
@@ -155,7 +164,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 +181,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/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,19 @@
 import logging
+import platform
 import random
 import re
 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]+')

patroni/version.py

@@ -1 +1 @@
-__version__ = '1.6.1'
+__version__ = '1.6.3'

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
 
 
@@ -183,7 +172,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,10 @@ 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

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

tests/test_kubernetes.py

@@ -157,17 +157,18 @@ class TestCacheBuilder(unittest.TestCase):
     @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

tests/test_postgresql.py

@@ -1,6 +1,7 @@
 import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
 import psycopg2
+import re
 import subprocess
 import time
 
@@ -220,7 +221,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()})
@@ -620,7 +622,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):

tests/test_postmaster.py

@@ -1,3 +1,4 @@
+import multiprocessing
 import psutil
 import unittest
 
@@ -96,6 +97,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):