Release v2.1.6 (#2507)

* bump version
* update release notes

Co-authored-by: Alexander Kukushkin <[email protected]>

.github/workflows/install_deps.py

@@ -5,7 +5,6 @@ import subprocess
 import stat
 import sys
 import tarfile
-import time
 import zipfile
 
 
@@ -47,13 +46,7 @@ def install_packages(what):
     packages = packages.get(what, [])
     ver = versions.get(what)
     subprocess.call(['sudo', 'apt-get', 'update', '-y'])
-    subprocess.call(['sudo', 'apt-get', 'install', '-y', 'wget', 'ca-certificates', 'gnupg', 'expect-dev'])
+    return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev'] + packages)
-    subprocess.call(['sudo', 'sh', '-c', "wget -qO - https://www.postgresql.org/media/keys/ACCC4CF8.asc"
-                     " | gpg --dearmor > /etc/apt/trusted.gpg.d/apt.postgresql.org.gpg"])
-    subprocess.call(['sudo', 'sed', '-i', 's/pgdg main.*$/pgdg main {0}/'.format(ver),
-                     '/etc/apt/sources.list.d/pgdg.list'])
-    subprocess.call(['sudo', 'apt-get', 'update', '-y'])
-    return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver] + packages)
 
 
 def get_file(url, name):
@@ -127,57 +120,12 @@ def install_postgres():
     return 0
 
 
-def setup_kubernetes():
-    get_file('https://storage.googleapis.com/minikube/k8sReleases/v1.7.0/localkube-linux-amd64', 'localkube')
-    chmod_755('localkube')
-
-    devnull = open(os.devnull, 'w')
-    subprocess.Popen(['sudo', 'nohup', './localkube', '--logtostderr=true', '--enable-dns=false'],
-                     stdout=devnull, stderr=devnull)
-    for _ in range(0, 120):
-        if subprocess.call(['wget', '-qO', '-', 'http://127.0.0.1:8080/'], stdout=devnull, stderr=devnull) == 0:
-            break
-        time.sleep(1)
-    else:
-        print('localkube did not start')
-        return 1
-
-    subprocess.call('sudo chmod 644 /var/lib/localkube/certs/*', shell=True)
-    print('Set up .kube/config')
-    kube = os.path.join(os.path.expanduser('~'), '.kube')
-    os.makedirs(kube)
-    with open(os.path.join(kube, 'config'), 'w') as f:
-        f.write("""apiVersion: v1
-clusters:
-- cluster:
-    certificate-authority: /var/lib/localkube/certs/ca.crt
-    server: https://127.0.0.1:8443
-  name: local
-contexts:
-- context:
-    cluster: local
-    user: myself
-  name: local
-current-context: local
-kind: Config
-preferences: {}
-users:
-- name: myself
-  user:
-    client-certificate: /var/lib/localkube/certs/apiserver.crt
-    client-key: /var/lib/localkube/certs/apiserver.key
-""")
-    return 0
-
-
 def main():
     what = os.environ.get('DCS', sys.argv[1] if len(sys.argv) > 1 else 'all')
 
     if what != 'all':
         if sys.platform.startswith('linux'):
             r = install_packages(what)
-            if r == 0 and what == 'kubernetes':
-                r = setup_kubernetes()
         else:
             r = install_postgres()
 

.github/workflows/run_tests.py

@@ -35,6 +35,8 @@ def main():
         unbuffer = []
     env['PATH'] = path + os.pathsep + env['PATH']
     env['DCS'] = what
+    if what == 'kubernetes':
+        env['PATRONI_KUBERNETES_CONTEXT'] = 'k3d-k3s-default'
 
     ret = subprocess.call(unbuffer + [sys.executable, '-m', 'behave'], env=env)
 

.github/workflows/tests.yaml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   unit:
-    runs-on: ${{ matrix.os }}-latest
+    runs-on: ${{ fromJson('{"ubuntu":"ubuntu-20.04","windows":"windows-latest","macos":"macos-latest"}')[matrix.os] }}
     strategy:
       fail-fast: false
       matrix:
@@ -89,7 +89,7 @@ jobs:
       run: python -m coveralls --service=github
 
   behave:
-    runs-on: ${{ matrix.os }}-latest
+    runs-on: ${{ fromJson('{"ubuntu":"ubuntu-20.04","windows":"windows-latest","macos":"macos-latest"}')[matrix.os] }}
     env:
       DCS: ${{ matrix.dcs }}
       ETCDVERSION: 3.3.13
@@ -120,8 +120,14 @@ jobs:
       uses: actions/setup-python@v4
       with:
         python-version: ${{ matrix.python-version }}
+    - uses: nolar/setup-k3d-k3s@v1
+      if: matrix.dcs == 'kubernetes'
     - name: Add postgresql apt repo
-      run: sudo sh -c 'echo "deb http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list'
+      run: |
+        sudo apt-get update -y
+        sudo apt-get install -y wget ca-certificates gnupg
+        sudo sh -c 'echo "deb http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list'
+        sudo sh -c 'wget -qO - https://www.postgresql.org/media/keys/ACCC4CF8.asc | gpg --dearmor > /etc/apt/trusted.gpg.d/apt.postgresql.org.gpg'
       if: matrix.os == 'ubuntu'
     - name: Install dependencies
       run: python .github/workflows/install_deps.py

Dockerfile

@@ -50,11 +50,11 @@ RUN set -ex \
     && chown -R postgres:postgres /var/log \
 \
     # Download etcd
-    && curl -sL https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz \
+    && curl -sL https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-$(dpkg --print-architecture).tar.gz \
             | tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl \
 \
     # Download confd
-    && curl -sL https://github.com/kelseyhightower/confd/releases/download/v${CONFDVERSION}/confd-${CONFDVERSION}-linux-amd64 \
+    && curl -sL https://github.com/kelseyhightower/confd/releases/download/v${CONFDVERSION}/confd-${CONFDVERSION}-linux-$(dpkg --print-architecture) \
             > /usr/local/bin/confd && chmod +x /usr/local/bin/confd \
 \
     # Clean up all useless packages and some files
@@ -90,7 +90,7 @@ RUN set -ex \
     && find /usr/bin -xtype l -delete \
     && find /var/log -type f -exec truncate --size 0 {} \; \
     && find /usr/lib/python3/dist-packages -name '*test*' | xargs rm -fr \
-    && find /lib/x86_64-linux-gnu/security -type f ! -name pam_env.so ! -name pam_permit.so ! -name pam_unix.so -delete
+    && find /lib/$(uname -m)-linux-gnu/security -type f ! -name pam_env.so ! -name pam_permit.so ! -name pam_unix.so -delete
 
 # perform compression if it is necessary
 ARG COMPRESS
@@ -99,8 +99,10 @@ RUN if [ "$COMPRESS" = "true" ]; then \
         # Allow certain sudo commands from postgres
         && echo 'postgres ALL=(ALL) NOPASSWD: /bin/tar xpJf /a.tar.xz -C /, /bin/rm /a.tar.xz, /bin/ln -snf dash /bin/sh' >> /etc/sudoers \
         && ln -snf busybox /bin/sh \
-        && files="/bin/sh /usr/bin/sudo /usr/lib/sudo/sudoers.so /lib/x86_64-linux-gnu/security/pam_*.so" \
+        && arch=$(uname -m) \
-        && libs="$(ldd $files | awk '{print $3;}' | grep '^/' | sort -u) /lib/x86_64-linux-gnu/ld-linux-x86-64.so.* /lib/x86_64-linux-gnu/libnsl.so.* /lib/x86_64-linux-gnu/libnss_compat.so.*" \
+        && darch=$(uname -m | sed 's/_/-/') \
+        && files="/bin/sh /usr/bin/sudo /usr/lib/sudo/sudoers.so /lib/$arch-linux-gnu/security/pam_*.so" \
+        && libs="$(ldd $files | awk '{print $3;}' | grep '^/' | sort -u) /lib/ld-linux-$darch.so.* /lib/$arch-linux-gnu/ld-linux-$darch.so.* /lib/$arch-linux-gnu/libnsl.so.* /lib/$arch-linux-gnu/libnss_compat.so.* /lib/$arch-linux-gnu/libnss_files.so.*" \
         && (echo /var/run $files $libs | tr ' ' '\n' && realpath $files $libs) | sort -u | sed 's/^\///' > /exclude \
         && find /etc/alternatives -xtype l -delete \
         && save_dirs="usr lib var bin sbin etc/ssl etc/init.d etc/alternatives etc/apt" \

docs/releases.rst

@@ -3,6 +3,61 @@
 Release notes
 =============
 
+Version 2.1.6
+-------------
+
+**Improvements**
+
+- Fix annoying exceptions on ssl socket shutdown (Alexander Kukushkin)
+
+  The HAProxy is closing connections as soon as it got the HTTP Status code leaving no time for Patroni to properly shutdown SSL connection.
+
+- Adjust example Dockerfile for arm64 (Polina Bungina)
+
+  Remove explicit ``amd64`` and ``x86_64``, don't remove ``libnss_files.so.*``.
+
+
+**Security improvements**
+
+- Enforce ``search_path=pg_catalog`` for non-replication connections (Alexander)
+
+  Since Patroni is heavily relying on superuser connections, we want to protect it from the possible attacks carried out using user-defined functions and/or operators in ``public`` schema with the same name and signature as the corresponding objects in ``pg_catalog``. For that, ``search_path=pg_catalog`` is enforced for all connections created by Patroni (except replication connections).
+
+- Prevent passwords from being recorded in ``pg_stat_statements`` (Feike Steenbergen)
+
+  It is achieved by setting ``pg_stat_statements.track_utility=off`` when creating users.
+
+
+**Bugfixes**
+
+- Declare ``proxy_address`` as optional (Denis Laxalde)
+
+  As it is effectively a non-required option.
+
+- Improve behaviour of the insecure option (Alexander)
+
+  Ctl's ``insecure`` option didn't work properly when client certificates were used for REST API requests.
+
+- Take watchdog configuration from ``bootstrap.dcs`` when the new cluster is bootstrapped (Matt Baker)
+
+  Patroni used to initially configure watchdog with defaults when bootstrapping a new cluster rather than taking configuration used to bootstrap the DCS.
+
+- Fix the way file extensions are treated while finding executables in WIN32 (Martín Marqués)
+
+  Only add ``.exe`` to a file name if it has no extension yet.
+
+- Fix Consul TTL setup (Alexander)
+
+  We used ``ttl/2.0`` when setting the value on the HTTPClient, but forgot to multiply the current value by 2 in the class' property. It was resulting in Consul TTL off by twice.
+
+
+**Removed functionality**
+
+- Remove ``patronictl configure`` (Polina)
+
+  There is no more need for a separate ``patronictl`` config creation.
+
+
 Version 2.1.5
 -------------
 

features/environment.py

@@ -2,6 +2,7 @@ import abc
 import datetime
 import os
 import json
+import re
 import shutil
 import signal
 import six
@@ -177,7 +178,7 @@ class PatroniController(AbstractController):
         patroni_config_name = self.PATRONI_CONFIG.format(name)
         patroni_config_path = os.path.join(self._output_dir, patroni_config_name)
 
-        with open(patroni_config_name) as f:
+        with open('postgres0.yml') as f:
             config = yaml.safe_load(f)
             config.pop('etcd', None)
 
@@ -186,8 +187,10 @@ class PatroniController(AbstractController):
             os.environ['RAFT_PORT'] = str(int(raft_port) + 1)
             config['raft'] = {'data_dir': self._output_dir, 'self_addr': 'localhost:' + os.environ['RAFT_PORT']}
 
-        host = config['postgresql']['listen'].split(':')[0]
+        host = config['restapi']['listen'].rsplit(':', 1)[0]
+        config['restapi']['listen'] = config['restapi']['connect_address'] = '{0}:{1}'.format(host, 8008+int(name[-1]))
 
+        host = config['postgresql']['listen'].rsplit(':', 1)[0]
         config['postgresql']['listen'] = config['postgresql']['connect_address'] = '{0}:{1}'.format(host, self.__PORT)
 
         config['name'] = name
@@ -200,7 +203,32 @@ class PatroniController(AbstractController):
             'logging_collector': 'on', 'log_destination': 'csvlog',
             'log_directory': self._output_dir.replace('\\', '/'),
             'log_filename': name + '.log', 'log_statement': 'all', 'log_min_messages': 'debug1',
-            'unix_socket_directories': tempfile.gettempdir().replace('\\', '/')})
+            'shared_buffers': '1MB', 'unix_socket_directories': tempfile.gettempdir().replace('\\', '/')})
+        config['postgresql']['pg_hba'] = [
+            'local all all trust',
+            'local replication all trust',
+            'host replication replicator all md5',
+            'host all all all md5'
+        ]
+
+        if self._context.postgres_supports_ssl and self._context.certfile:
+            config['postgresql']['parameters'].update({
+                'ssl': 'on',
+                'ssl_ca_file': self._context.certfile.replace('\\', '/'),
+                'ssl_cert_file': self._context.certfile.replace('\\', '/'),
+                'ssl_key_file': self._context.keyfile.replace('\\', '/')
+            })
+            for user in config['postgresql'].get('authentication').keys():
+                config['postgresql'].get('authentication', {}).get(user, {}).update({
+                    'sslmode': 'verify-ca',
+                    'sslrootcert': self._context.certfile,
+                    'sslcert': self._context.certfile,
+                    'sslkey': self._context.keyfile
+                })
+            for i, line in enumerate(list(config['postgresql']['pg_hba'])):
+                if line.endswith('md5'):
+                    # we want to verify client cert first and than password
+                    config['postgresql']['pg_hba'][i] = 'hostssl' + line[4:] + ' clientcert=verify-ca'
 
         if 'bootstrap' in config:
             config['bootstrap']['post_bootstrap'] = 'psql -w -c "SELECT 1"'
@@ -218,20 +246,21 @@ class PatroniController(AbstractController):
         with open(patroni_config_path, 'w') as f:
             yaml.safe_dump(config, f, default_flow_style=False)
 
-        user = config['postgresql'].get('authentication', config['postgresql']).get('superuser', {})
+        self._connkwargs = config['postgresql'].get('authentication', config['postgresql']).get('superuser', {})
-        self._connkwargs = {k: user[n] for n, k in [('username', 'user'), ('password', 'password')] if n in user}
+        self._connkwargs.update({'host': host, 'port': self.__PORT, 'dbname': 'postgres',
-        self._connkwargs.update({'host': host, 'port': self.__PORT, 'dbname': 'postgres'})
+                                 'user': self._connkwargs.pop('username', None)})
 
         self._replication = config['postgresql'].get('authentication', config['postgresql']).get('replication', {})
-        self._replication.update({'host': host, 'port': self.__PORT, 'dbname': 'postgres'})
+        self._replication.update({'host': host, 'port': self.__PORT, 'user': self._replication.pop('username', None)})
         self._restapi_url = 'http://{0}'.format(config['restapi']['connect_address'])
+        if self._context.certfile:
+            self._restapi_url = self._restapi_url.replace('http://', 'https://')
 
         return patroni_config_path
 
     def _connection(self):
         if not self._conn or self._conn.closed != 0:
             self._conn = psycopg.connect(**self._connkwargs)
-            self._conn.autocommit = True
         return self._conn
 
     def _cursor(self):
@@ -284,7 +313,10 @@ class PatroniController(AbstractController):
 
     @property
     def backup_source(self):
-        return 'postgres://{username}:{password}@{host}:{port}/{dbname}'.format(**self._replication)
+        def escape(value):
+            return re.sub(r'([\'\\ ])', r'\\\1', str(value))
+
+        return ' '.join('{0}={1}'.format(k, escape(v)) for k, v in self._replication.items())
 
     def backup(self, dest=os.path.join('data', 'basebackup')):
         subprocess.call(PatroniPoolController.BACKUP_SCRIPT + ['--walmethod=none',
@@ -476,10 +508,10 @@ class KubernetesController(AbstractDcsController):
         self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items())
         os.environ['PATRONI_KUBERNETES_LABELS'] = json.dumps(self._labels)
         os.environ['PATRONI_KUBERNETES_USE_ENDPOINTS'] = 'true'
-        os.environ['PATRONI_KUBERNETES_BYPASS_API_SERVICE'] = 'true'
+        os.environ.setdefault('PATRONI_KUBERNETES_BYPASS_API_SERVICE', 'true')
 
         from patroni.dcs.kubernetes import k8s_client, k8s_config
-        k8s_config.load_kube_config(context='local')
+        k8s_config.load_kube_config(context=os.environ.setdefault('PATRONI_KUBERNETES_CONTEXT', 'kind-kind'))
         self._client = k8s_client
         self._api = self._client.CoreV1Api()
 
@@ -660,8 +692,17 @@ class PatroniPoolController(object):
         self._patroni_path = None
         self._processes = {}
         self.create_and_set_output_directory('')
+        self._check_postgres_ssl()
         self.known_dcs = {subclass.name(): subclass for subclass in AbstractDcsController.get_subclasses()}
 
+    def _check_postgres_ssl(self):
+        try:
+            subprocess.check_output(['postgres', '-D', os.devnull, '-c', 'ssl=on'], stderr=subprocess.STDOUT)
+            raise Exception  # this one should never happen because the previous line will always raise and exception
+        except Exception as e:
+            self._context.postgres_supports_ssl = isinstance(e, subprocess.CalledProcessError)\
+                    and 'SSL is not supported by this build' not in e.output.decode()
+
     @property
     def patroni_path(self):
         if self._patroni_path is None:
@@ -712,7 +753,8 @@ class PatroniPoolController(object):
             'bootstrap': {
                 'method': 'pg_basebackup',
                 'pg_basebackup': {
-                    'command': " ".join(self.BACKUP_SCRIPT) + ' --walmethod=stream --dbname=' + f.backup_source
+                    'command': " ".join(self.BACKUP_SCRIPT +
+                                        ['--walmethod=stream', '--dbname="{0}"'.format(f.backup_source)])
                 },
                 'dcs': {
                     'postgresql': {
@@ -889,13 +931,32 @@ class WatchdogMonitor(object):
 
 # actions to execute on start/stop of the tests and before running individual features
 def before_all(context):
-    os.environ.update({'PATRONI_RESTAPI_USERNAME': 'username', 'PATRONI_RESTAPI_PASSWORD': 'password'})
-    context.request_executor = PatroniRequest({'ctl': {'auth': os.environ['PATRONI_RESTAPI_USERNAME'] +
-                                               ':' + os.environ['PATRONI_RESTAPI_PASSWORD']}})
     context.ci = os.name == 'nt' or\
         any(a in os.environ for a in ('TRAVIS_BUILD_NUMBER', 'BUILD_NUMBER', 'GITHUB_ACTIONS'))
     context.timeout_multiplier = 5 if context.ci else 1  # MacOS sometimes is VERY slow
     context.pctl = PatroniPoolController(context)
+
+    context.keyfile = os.path.join(context.pctl.output_dir, 'patroni.key')
+    context.certfile = os.path.join(context.pctl.output_dir, 'patroni.crt')
+    try:
+        with open(os.devnull, 'w') as null:
+            ret = subprocess.call(['openssl', 'req', '-nodes', '-new', '-x509', '-subj', '/CN=batman.patroni',
+                                   '-keyout', context.keyfile, '-out', context.certfile], stdout=null, stderr=null)
+            if ret != 0:
+                raise Exception
+    except Exception:
+        context.keyfile = context.certfile = None
+
+    os.environ.update({'PATRONI_RESTAPI_USERNAME': 'username', 'PATRONI_RESTAPI_PASSWORD': 'password'})
+    ctl = {'auth': os.environ['PATRONI_RESTAPI_USERNAME'] + ':' + os.environ['PATRONI_RESTAPI_PASSWORD']}
+    if context.certfile:
+        os.environ.update({'PATRONI_RESTAPI_CAFILE': context.certfile,
+                           'PATRONI_RESTAPI_CERTFILE': context.certfile,
+                           'PATRONI_RESTAPI_KEYFILE': context.keyfile,
+                           'PATRONI_RESTAPI_VERIFY_CLIENT': 'required',
+                           'PATRONI_CTL_INSECURE': 'on'})
+        ctl.update({'cacert': context.certfile, 'certfile': context.certfile, 'keyfile': context.keyfile})
+    context.request_executor = PatroniRequest({'ctl': ctl}, True)
     context.dcs_ctl = context.pctl.known_dcs[context.pctl.dcs](context)
     context.dcs_ctl.start()
     try:

features/steps/basic_replication.py

@@ -28,7 +28,7 @@ def stop_postgres(context, name):
 def add_table(context, table_name, pg_name):
     # parse the configuration file and get the port
     try:
-        context.pctl.query(pg_name, "CREATE TABLE {0}()".format(table_name))
+        context.pctl.query(pg_name, "CREATE TABLE public.{0}()".format(table_name))
     except pg.Error as e:
         assert False, "Error creating table {0} on {1}: {2}".format(table_name, pg_name, e)
 
@@ -37,9 +37,9 @@ def add_table(context, table_name, pg_name):
 def toggle_wal_replay(context, action, pg_name):
     # pause or resume the wal replay process
     try:
-       version = context.pctl.query(pg_name, "select pg_catalog.pg_read_file('PG_VERSION', 0, 2)").fetchone()
+        version = context.pctl.query(pg_name, "SHOW server_version_num").fetchone()[0]
-       wal = version and version[0] and int(version[0].split('.')[0]) < 10 and "xlog" or "wal"
+        wal_name = 'xlog' if int(version)/10000 < 10 else 'wal'
-       context.pctl.query(pg_name, "SELECT pg_{0}_replay_{1}()".format(wal, action))
+        context.pctl.query(pg_name, "SELECT pg_{0}_replay_{1}()".format(wal_name, action))
     except pg.Error as e:
         assert False, "Error during {0} wal recovery on {1}: {2}".format(action, pg_name, e)
 
@@ -47,10 +47,10 @@ def toggle_wal_replay(context, action, pg_name):
 @step('I {action:w} table on {pg_name:w}')
 def crdr_mytest(context, action, pg_name):
     try:
-       if (action == "create"):
+        if (action == "create"):
-           context.pctl.query(pg_name, "create table if not exists mytest(id Numeric)")
+            context.pctl.query(pg_name, "create table if not exists public.mytest(id numeric)")
-       else:
+        else:
-           context.pctl.query(pg_name, "drop table if exists mytest")
+            context.pctl.query(pg_name, "drop table if exists public.mytest")
     except pg.Error as e:
         assert False, "Error {0} table mytest on {1}: {2}".format(action, pg_name, e)
 
@@ -59,7 +59,7 @@ def crdr_mytest(context, action, pg_name):
 def initiate_load(context, pg_name):
     # perform dummy load
     try:
-       context.pctl.query(pg_name, "begin; insert into mytest select r::numeric from generate_series(1, 350000) r; commit;")
+        context.pctl.query(pg_name, "insert into public.mytest select r::numeric from generate_series(1, 350000) r")
     except pg.Error as e:
         assert False, "Error loading test data on {0}: {1}".format(pg_name, e)
 
@@ -68,7 +68,7 @@ def initiate_load(context, pg_name):
 def table_is_present_on(context, table_name, pg_name, max_replication_delay):
     max_replication_delay *= context.timeout_multiplier
     for _ in range(int(max_replication_delay)):
-        if context.pctl.query(pg_name, "SELECT 1 FROM {0}".format(table_name), fail_ok=True) is not None:
+        if context.pctl.query(pg_name, "SELECT 1 FROM public.{0}".format(table_name), fail_ok=True) is not None:
             break
         sleep(1)
     else:

features/steps/patroni_api.py

@@ -1,5 +1,4 @@
 import json
-import os
 import parse
 import shlex
 import subprocess
@@ -71,6 +70,8 @@ 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):
+    if context.certfile:
+        url = url.replace('http://', 'https://')
     data = data and json.loads(data)
     try:
         r = context.request_executor.request(request_method, url, data)
@@ -86,10 +87,7 @@ def do_request(context, request_method, url, data):
 def do_run(context, cmd):
     cmd = [sys.executable, '-m', 'coverage', 'run', '--source=patroni', '-p'] + shlex.split(cmd)
     try:
-        # XXX: Dirty hack! We need to take name/passwd from the config!
+        response = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
-        env = os.environ.copy()
-        env.update({'PATRONI_RESTAPI_USERNAME': 'username', 'PATRONI_RESTAPI_PASSWORD': 'password'})
-        response = subprocess.check_output(cmd, stderr=subprocess.STDOUT, env=env)
         context.status_code = 0
     except subprocess.CalledProcessError as e:
         response = e.output
@@ -135,6 +133,8 @@ def add_tag_to_config(context, tag, value, pg_name):
 
 @then('Response on GET {url} contains {value} after {timeout:d} seconds')
 def check_http_response(context, url, value, timeout, negate=False):
+    if context.certfile:
+        url = url.replace('http://', 'https://')
     timeout *= context.timeout_multiplier
     for _ in range(int(timeout)):
         r = context.request_executor.request('GET', url)

features/steps/watchdog.py

@@ -15,7 +15,7 @@ def polling_loop(timeout, interval=1):
 
 @step('I start {name:w} with watchdog')
 def start_patroni_with_watchdog(context, name):
-    return context.pctl.start(name, custom_config={'watchdog': True})
+    return context.pctl.start(name, custom_config={'watchdog': True, 'bootstrap': {'dcs': {'ttl': 20}}})
 
 
 @step('{name:w} watchdog has been pinged after {timeout:d} seconds')
@@ -31,6 +31,11 @@ def watchdog_was_closed(context, name):
     assert context.pctl.get_watchdog(name).was_closed
 
 
+@step('{name:w} watchdog has a {timeout:d} second timeout')
+def watchdog_has_timeout(context, name, timeout):
+    assert context.pctl.get_watchdog(name).timeout == timeout
+
+
 @step('I reset {name:w} watchdog state')
 def watchdog_reset_pinged(context, name):
     context.pctl.get_watchdog(name).reset()

features/watchdog.feature

@@ -6,6 +6,14 @@ Feature: watchdog
     Then postgres0 is a leader after 10 seconds
     And postgres0 role is the primary after 10 seconds
     And postgres0 watchdog has been pinged after 10 seconds
+    And postgres0 watchdog has a 15 second timeout
+
+  Scenario: watchdog is reconfigured after global ttl changed
+    Given I run patronictl.py edit-config batman -s ttl=30 --force
+    Then I receive a response returncode 0
+    And I receive a response output "+ttl: 30"
+    When I sleep for 4 seconds
+    Then postgres0 watchdog has a 25 second timeout
 
   Scenario: watchdog is disabled during pause
     Given I run patronictl.py pause batman

patroni/__main__.py

@@ -47,6 +47,7 @@ class Patroni(AbstractPatroniDaemon):
                 elif not self.config.dynamic_configuration and 'bootstrap' in self.config:
                     if self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']):
                         self.dcs.reload_config(self.config)
+                        self.watchdog.reload_config(self.config)
                 break
             except DCSError:
                 logger.warning('Can not get cluster from dcs')

patroni/api.py

@@ -629,7 +629,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             stmt = ("SELECT " + postgresql.POSTMASTER_START_TIME + ", " + postgresql.TL_LSN + ","
                     " pg_catalog.pg_last_xact_replay_timestamp(),"
                     " pg_catalog.array_to_json(pg_catalog.array_agg(pg_catalog.row_to_json(ri))) "
-                    "FROM (SELECT (SELECT rolname FROM pg_authid WHERE oid = usesysid) AS usename,"
+                    "FROM (SELECT (SELECT rolname FROM pg_catalog.pg_authid WHERE oid = usesysid) AS usename,"
                     " application_name, client_addr, w.state, sync_state, sync_priority"
                     " FROM pg_catalog.pg_stat_get_wal_senders() w, pg_catalog.pg_stat_get_activity(pid)) AS ri")
 
@@ -845,7 +845,10 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
 
     def shutdown_request(self, request):
         if hasattr(request, 'context'):  # SSLSocket
-            request.unwrap()
+            try:
+                request.unwrap()
+            except Exception as e:
+                logger.debug('Failed to shutdown SSL connection: %r', e)
         super(RestApiServer, self).shutdown_request(request)
 
     def get_certificate_serial_number(self):

patroni/ctl.py

@@ -110,7 +110,7 @@ def parse_dcs(dcs):
     return yaml.safe_load(default['template'].format(host=parsed.hostname or 'localhost', port=port or default['port']))
 
 
-def load_config(path, dcs):
+def load_config(path, dcs_url):
     from patroni.config import Config
 
     if not (os.path.exists(path) and os.access(path, os.R_OK)):
@@ -123,22 +123,14 @@ def load_config(path, dcs):
         logging.debug('Loading configuration from file %s', path)
     config = Config(path, validator=None).copy()
 
-    dcs = parse_dcs(dcs) or parse_dcs(config.get('dcs_api')) or {}
+    dcs_url = parse_dcs(dcs_url) or {}
-    if dcs:
+    if dcs_url:
         for d in DCS_DEFAULTS:
             config.pop(d, None)
-        config.update(dcs)
+        config.update(dcs_url)
     return config
 
 
-def store_config(config, path):
-    dir_path = os.path.dirname(path)
-    if dir_path and not os.path.isdir(dir_path):
-        os.makedirs(dir_path)
-    with open(path, 'w') as fd:
-        yaml.dump(config, fd)
-
-
 option_format = click.option('--format', '-f', 'fmt', help='Output format (pretty, tsv, json, yaml)', default='pretty')
 option_watchrefresh = click.option('-w', '--watch', type=float, help='Auto update the screen every X seconds')
 option_watch = click.option('-W', is_flag=True, help='Auto update the screen every 2 seconds')
@@ -151,16 +143,16 @@ option_insecure = click.option('-k', '--insecure', is_flag=True, help='Allow con
 @click.group()
 @click.option('--config-file', '-c', help='Configuration file',
               envvar='PATRONICTL_CONFIG_FILE', default=CONFIG_FILE_PATH)
-@click.option('--dcs', '-d', help='Use this DCS', envvar='DCS')
+@click.option('--dcs-url', '--dcs', '-d', help='The DCS connect url', envvar='DCS_URL')
 @option_insecure
 @click.pass_context
-def ctl(ctx, config_file, dcs, insecure):
+def ctl(ctx, config_file, dcs_url, insecure):
     level = 'WARNING'
     for name in ('LOGLEVEL', 'PATRONI_LOGLEVEL', 'PATRONI_LOG_LEVEL'):
         level = os.environ.get(name, level)
     logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=level)
     logging.captureWarnings(True)  # Capture eventual SSL warning
-    ctx.obj = load_config(config_file, dcs)
+    ctx.obj = load_config(config_file, dcs_url)
     # backward compatibility for configuration file where ctl section is not define
     ctx.obj.setdefault('ctl', {})['insecure'] = ctx.obj.get('ctl', {}).get('insecure') or insecure
 
@@ -282,7 +274,6 @@ def get_cursor(cluster, connect_parameters, role='master', member=None):
 
     from . import psycopg
     conn = psycopg.connect(**params)
-    conn.autocommit = True
     cursor = conn.cursor()
     if role == 'any':
         return cursor
@@ -897,14 +888,6 @@ def timestamp(precision=6):
     return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:precision - 7]
 
 
-@ctl.command('configure', help='Create configuration file')
-@click.option('--config-file', '-c', help='Configuration file', prompt='Configuration file', default=CONFIG_FILE_PATH)
-@click.option('--dcs', '-d', help='The DCS connect url', prompt='DCS connect url', default='etcd://localhost:2379')
-@click.option('--namespace', '-n', help='The namespace', prompt='Namespace', default='/service/')
-def configure(config_file, dcs, namespace):
-    store_config({'dcs_api': str(dcs), 'namespace': str(namespace)}, config_file)
-
-
 def touch_member(config, dcs):
     ''' Rip-off of the ha.touch_member without inter-class dependencies '''
     p = Postgresql(config['postgresql'])

patroni/dcs/__init__.py

@@ -444,7 +444,7 @@ class TimelineHistory(namedtuple('TimelineHistory', 'index,value,lines')):
         return TimelineHistory(index, value, lines)
 
 
-class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,failover,sync,history,slots')):
+class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,failover,sync,history,slots,failsafe')):
 
     """Immutable object (namedtuple) which represents PostgreSQL cluster.
     Consists of the following fields:
@@ -606,11 +606,11 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,f
     @property
     def timeline(self):
         """
-        >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0, 0).timeline
+        >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0, 0, None).timeline
         0
-        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]'), 0).timeline
+        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]'), 0, None).timeline
         1
-        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]'), 0).timeline
+        >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]'), 0, None).timeline
         0
         """
         if self.history:
@@ -628,6 +628,20 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,f
         return next(iter(sorted(filter(lambda v: v, [m.version for m in self.members])) + [None]))
 
 
+class ReturnFalseException(Exception):
+    pass
+
+
+def catch_return_false_exception(func):
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except ReturnFalseException:
+            return False
+
+    return wrapper
+
+
 @six.add_metaclass(abc.ABCMeta)
 class AbstractDCS(object):
 
@@ -641,6 +655,7 @@ class AbstractDCS(object):
     _STATUS = 'status'  # JSON, contains "leader_lsn" and confirmed_flush_lsn of logical "slots" on the leader
     _LEADER_OPTIME = _OPTIME + '/' + _LEADER  # legacy
     _SYNC = 'sync'
+    _FAILSAFE = 'failsafe'
 
     def __init__(self, config):
         """
@@ -658,6 +673,7 @@ class AbstractDCS(object):
         self._last_lsn = ''
         self._last_seen = 0
         self._last_status = {}
+        self._last_failsafe = {}
         self.event = Event()
 
     def client_path(self, path):
@@ -703,6 +719,10 @@ class AbstractDCS(object):
     def sync_path(self):
         return self.client_path(self._SYNC)
 
+    @property
+    def failsafe_path(self):
+        return self.client_path(self._FAILSAFE)
+
     @abc.abstractmethod
     def set_ttl(self, ttl):
         """Set the new ttl value for leader key"""
@@ -754,6 +774,8 @@ class AbstractDCS(object):
             raise
 
         self._last_seen = int(time.time())
+        self._last_status = {self._OPTIME: cluster.last_lsn, 'slots': cluster.slots}
+        self._last_failsafe = cluster.failsafe
 
         with self._cluster_thread_lock:
             self._cluster = cluster
@@ -796,23 +818,35 @@ class AbstractDCS(object):
             self._last_lsn = value[self._OPTIME]
             self._write_leader_optime(str(value[self._OPTIME]))
 
+    @abc.abstractmethod
+    def _write_failsafe(self, value):
+        """Write current cluster topology to DCS that will be used by failsafe mechanism (if enabled).
+
+        :param value: failsafe topology serialized in JSON format
+        :returns: `!True` on success."""
+
+    def write_failsafe(self, value):
+        if not (isinstance(self._last_failsafe, dict) and deep_compare(self._last_failsafe, value))\
+                and self._write_failsafe(json.dumps(value, separators=(',', ':'))):
+            self._last_failsafe = value
+
     @abc.abstractmethod
     def _update_leader(self):
         """Update leader key (or session) ttl
 
         :returns: `!True` if leader key (or session) has been updated successfully.
-            If not, `!False` must be returned and current instance would be demoted.
 
         You have to use CAS (Compare And Swap) operation in order to update leader key,
-        for example for etcd `prevValue` parameter must be used."""
+        for example for etcd `prevValue` parameter must be used.
+        If update fails due to DCS not being accessible or because it is not able to
+        process requests (hopefuly temporary), the ~DCSError exception should be raised."""
 
-    def update_leader(self, last_lsn, slots=None):
+    def update_leader(self, last_lsn, slots=None, failsafe=None):
         """Update leader key (or session) ttl and optime/leader
 
         :param last_lsn: absolute WAL LSN in bytes
         :param slots: dict with permanent slots confirmed_flush_lsn
-        :returns: `!True` if leader key (or session) has been updated successfully.
+        :returns: `!True` if leader key (or session) has been updated successfully."""
-            If not, `!False` must be returned and current instance would be demoted."""
 
         ret = self._update_leader()
         if ret and last_lsn:
@@ -820,6 +854,10 @@ class AbstractDCS(object):
             if slots:
                 status['slots'] = slots
             self.write_status(status)
+
+        if ret and failsafe is not None:
+            self.write_failsafe(failsafe)
+
         return ret
 
     @abc.abstractmethod
@@ -831,7 +869,10 @@ class AbstractDCS(object):
         :returns: `!True` if key has been created successfully.
 
         Key must be created atomically. In case if key already exists it should not be
-        overwritten and `!False` must be returned"""
+        overwritten and `!False` must be returned.
+
+        If key creation fails due to DCS not being accessible or because it is not able to
+        process requests (hopefuly temporary), the ~DCSError exception should be raised"""
 
     @abc.abstractmethod
     def set_failover_value(self, value, index=None):

patroni/dcs/consul.py

@@ -14,7 +14,8 @@ 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 . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member,\
+        SyncState, TimelineHistory, ReturnFalseException, catch_return_false_exception
 from ..exceptions import DCSError
 from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT
 
@@ -271,7 +272,7 @@ class Consul(AbstractDCS):
 
     @property
     def ttl(self):
-        return self._client.http.ttl
+        return self._client.http.ttl * 2  # we multiply the value by 2 because it was divided in the `set_ttl()` method
 
     def set_retry_timeout(self, retry_timeout):
         self._retry.deadline = retry_timeout
@@ -286,9 +287,9 @@ class Consul(AbstractDCS):
         except Exception:
             logger.exception('adjust_ttl')
 
-    def _do_refresh_session(self):
+    def _do_refresh_session(self, force=False):
         """:returns: `!True` if it had to create new session"""
-        if self._session and self._last_session_refresh + self._loop_wait > time.time():
+        if not force and self._session and self._last_session_refresh + self._loop_wait > time.time():
             return False
 
         if self._session:
@@ -373,11 +374,6 @@ class Consul(AbstractDCS):
 
             # get leader
             leader = nodes.get(self._LEADER)
-            if not self._ctl and leader and leader['Value'] == self._name \
-                    and self._session != leader.get('Session', 'x'):
-                logger.info('I am leader but not owner of the session. Removing leader node')
-                self._client.kv.delete(self.leader_path, cas=leader['ModifyIndex'])
-                leader = None
 
             if leader:
                 member = Member(-1, leader['Value'], None, {})
@@ -393,9 +389,16 @@ class Consul(AbstractDCS):
             sync = nodes.get(self._SYNC)
             sync = SyncState.from_node(sync and sync['ModifyIndex'], sync and sync['Value'])
 
-            return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
+            # get failsafe topology
+            failsafe = nodes.get(self._FAILSAFE)
+            try:
+                failsafe = json.loads(failsafe['Value']) if failsafe else None
+            except Exception:
+                failsafe = None
+
+            return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
         except NotFound:
-            return Cluster(None, None, None, None, [], None, None, None, None)
+            return Cluster(None, None, None, None, [], None, None, None, None, None)
         except Exception:
             logger.exception('get_cluster')
             raise ConsulError('Consul is not responding properly')
@@ -506,22 +509,34 @@ class Consul(AbstractDCS):
         ):
             return self._update_service(new_data)
 
-    @catch_consul_errors
+    def _do_attempt_to_acquire_leader(self, permanent, retry):
-    def _do_attempt_to_acquire_leader(self, permanent):
         try:
             kwargs = {} if permanent else {'acquire': self._session}
-            return self.retry(self._client.kv.put, self.leader_path, self._name, **kwargs)
+            return retry(self._client.kv.put, self.leader_path, self._name, **kwargs)
         except InvalidSession:
-            self._session = None
             logger.error('Our session disappeared from Consul. Will try to get a new one and retry attempt')
-            self.refresh_session()
+            self._session = None
-            return self.retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session)
+            retry.deadline = retry.stoptime - time.time()
 
+            retry(self._do_refresh_session)
+
+            retry.deadline = retry.stoptime - time.time()
+            if retry.deadline < 1:
+                raise ConsulError('_do_attempt_to_acquire_leader timeout')
+
+            return retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session)
+
+    @catch_return_false_exception
     def attempt_to_acquire_leader(self, permanent=False):
-        if not self._session and not permanent:
+        retry = self._retry.copy()
-            self.refresh_session()
+        if not permanent:
+            self._run_and_handle_exceptions(self._do_refresh_session, retry=retry)
 
-        ret = self._do_attempt_to_acquire_leader(permanent)
+            retry.deadline = retry.stoptime - time.time()
+            if retry.deadline < 1:
+                raise ConsulError('attempt_to_acquire_leader timeout')
+
+        ret = self._run_and_handle_exceptions(self._do_attempt_to_acquire_leader, permanent, retry, retry=None)
         if not ret:
             logger.info('Could not take out TTL lock')
 
@@ -547,10 +562,42 @@ class Consul(AbstractDCS):
         return self._client.kv.put(self.status_path, value)
 
     @catch_consul_errors
+    def _write_failsafe(self, value):
+        return self._client.kv.put(self.failsafe_path, value)
+
+    @staticmethod
+    def _run_and_handle_exceptions(method, *args, **kwargs):
+        retry = kwargs.pop('retry', None)
+        try:
+            return retry(method, *args, **kwargs) if retry else method(*args, **kwargs)
+        except (RetryFailedError, InvalidSession, HTTPException, HTTPError, socket.error, socket.timeout) as e:
+            raise ConsulError(e)
+        except ConsulException:
+            raise ReturnFalseException
+
+    @catch_return_false_exception
     def _update_leader(self):
+        retry = self._retry.copy()
+
+        self._run_and_handle_exceptions(self._do_refresh_session, True, retry=retry)
+
         if self._session:
-            self.retry(self._client.session.renew, self._session)
+            cluster = self.cluster
-            self._last_session_refresh = time.time()
+            leader_session = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session
+            if leader_session != self._session:
+                retry.deadline = retry.stoptime - time.time()
+                if retry.deadline < 1:
+                    raise ConsulError('update_leader timeout')
+                logger.warning('Recreating the leader key due to session mismatch')
+                if cluster.leader:
+                    self._run_and_handle_exceptions(self._client.kv.delete, self.leader_path, cas=cluster.leader.index)
+
+                retry.deadline = retry.stoptime - time.time()
+                if retry.deadline < 0.5:
+                    raise ConsulError('update_leader timeout')
+                self._run_and_handle_exceptions(self._client.kv.put, self.leader_path,
+                                                self._name, acquire=self._session)
+
         return bool(self._session)
 
     @catch_consul_errors

patroni/dcs/etcd.py

@@ -19,7 +19,8 @@ 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 . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member,\
+        SyncState, TimelineHistory, ReturnFalseException, catch_return_false_exception
 from ..exceptions import DCSError
 from ..request import get as requests_get
 from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT
@@ -460,6 +461,18 @@ class AbstractEtcd(AbstractDCS):
         if isinstance(raise_ex, Exception):
             raise raise_ex
 
+    def _run_and_handle_exceptions(self, method, *args, **kwargs):
+        retry = kwargs.pop('retry', self.retry)
+        try:
+            return retry(method, *args, **kwargs) if retry else method(*args, **kwargs)
+        except (RetryFailedError, etcd.EtcdConnectionFailed) as e:
+            raise self._client.ERROR_CLS(e)
+        except etcd.EtcdException as e:
+            self._handle_exception(e)
+            raise ReturnFalseException
+        except Exception as e:
+            self._handle_exception(e, raise_ex=self._client.ERROR_CLS('unexpected error'))
+
     @staticmethod
     def set_socket_options(sock, socket_options):
         if socket_options:
@@ -648,9 +661,16 @@ class Etcd(AbstractEtcd):
             sync = nodes.get(self._SYNC)
             sync = SyncState.from_node(sync and sync.modifiedIndex, sync and sync.value)
 
-            cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
+            # get failsafe topology
+            failsafe = nodes.get(self._FAILSAFE)
+            try:
+                failsafe = json.loads(failsafe.value) if failsafe else None
+            except Exception:
+                failsafe = None
+
+            cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
         except etcd.EtcdKeyNotFound:
-            cluster = Cluster(None, None, None, None, [], None, None, None, None)
+            cluster = Cluster(None, None, None, None, [], None, None, None, None, None)
         except Exception as e:
             self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
         self._has_failed = False
@@ -665,7 +685,7 @@ class Etcd(AbstractEtcd):
     def take_leader(self):
         return self.retry(self._client.write, self.leader_path, self._name, ttl=self._ttl)
 
-    def attempt_to_acquire_leader(self, permanent=False):
+    def _do_attempt_to_acquire_leader(self, permanent=False):
         try:
             return bool(self.retry(self._client.write,
                                    self.leader_path,
@@ -674,9 +694,11 @@ class Etcd(AbstractEtcd):
                                    prevExist=False))
         except etcd.EtcdAlreadyExist:
             logger.info('Could not take out TTL lock')
-        except (RetryFailedError, etcd.EtcdException):
+            return False
-            pass
+
-        return False
+    @catch_return_false_exception
+    def attempt_to_acquire_leader(self, permanent=False):
+        return self._run_and_handle_exceptions(self._do_attempt_to_acquire_leader, permanent=permanent, retry=None)
 
     @catch_etcd_errors
     def set_failover_value(self, value, index=None):
@@ -694,9 +716,20 @@ class Etcd(AbstractEtcd):
     def _write_status(self, value):
         return self._client.set(self.status_path, value)
 
+    def _do_update_leader(self):
+        try:
+            return self.retry(self._client.write, self.leader_path, self._name,
+                              prevValue=self._name, ttl=self._ttl) is not None
+        except etcd.EtcdKeyNotFound:
+            return self._do_attempt_to_acquire_leader()
+
     @catch_etcd_errors
+    def _write_failsafe(self, value):
+        return self._client.set(self.failsafe_path, value)
+
+    @catch_return_false_exception
     def _update_leader(self):
-        return self.retry(self._client.write, self.leader_path, self._name, prevValue=self._name, ttl=self._ttl)
+        return self._run_and_handle_exceptions(self._do_update_leader, retry=None)
 
     @catch_etcd_errors
     def initialize(self, create_new=True, sysid=""):

patroni/dcs/etcd3.py

@@ -13,7 +13,8 @@ import urllib3
 from threading import Condition, Lock, Thread
 from urllib3.exceptions import ReadTimeoutError, ProtocolError
 
-from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
+from . import ClusterConfig, Cluster, Failover, Leader, Member,\
+        SyncState, TimelineHistory, ReturnFalseException, catch_return_false_exception
 from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors
 from ..exceptions import DCSError, PatroniException
 from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT
@@ -609,8 +610,8 @@ class Etcd3(AbstractEtcd):
         if self.__do_not_watch:
             self._lease = None
 
-    def _do_refresh_lease(self, retry=None):
+    def _do_refresh_lease(self, force=False, retry=None):
-        if self._lease and self._last_lease_refresh + self._loop_wait > time.time():
+        if not force and self._lease and self._last_lease_refresh + self._loop_wait > time.time():
             return False
 
         if self._lease and not self._client.lease_keepalive(self._lease, retry):
@@ -711,7 +712,14 @@ class Etcd3(AbstractEtcd):
             sync = nodes.get(self._SYNC)
             sync = SyncState.from_node(sync and sync['mod_revision'], sync and sync['value'])
 
-            cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
+            # get failsafe topology
+            failsafe = nodes.get(self._FAILSAFE)
+            try:
+                failsafe = json.loads(failsafe['value']) if failsafe else None
+            except Exception:
+                failsafe = None
+
+            cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
         except UnsupportedEtcdVersion:
             raise
         except Exception as e:
@@ -744,21 +752,42 @@ class Etcd3(AbstractEtcd):
     def take_leader(self):
         return self.retry(self._client.put, self.leader_path, self._name, self._lease)
 
-    @catch_etcd_errors
+    def _do_attempt_to_acquire_leader(self, permanent, retry):
-    def _do_attempt_to_acquire_leader(self, permanent):
+        def _retry(*args, **kwargs):
+            kwargs['retry'] = retry
+            return retry(*args, **kwargs)
+
         try:
-            return self.retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0)
+            return _retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0)
         except LeaseNotFound:
-            self._lease = None
             logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt')
-            self.refresh_lease()
+            self._lease = None
-            return self.retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0)
+            retry.deadline = retry.stoptime - time.time()
 
+            _retry(self._do_refresh_lease)
+
+            retry.deadline = retry.stoptime - time.time()
+            if retry.deadline < 1:
+                raise Etcd3Error('_do_attempt_to_acquire_leader timeout')
+
+            return _retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0)
+
+    @catch_return_false_exception
     def attempt_to_acquire_leader(self, permanent=False):
-        if not self._lease and not permanent:
+        retry = self._retry.copy()
-            self.refresh_lease()
 
-        ret = self._do_attempt_to_acquire_leader(permanent)
+        def _retry(*args, **kwargs):
+            kwargs['retry'] = retry
+            return retry(*args, **kwargs)
+
+        if not permanent:
+            self._run_and_handle_exceptions(self._do_refresh_lease, retry=_retry)
+
+            retry.deadline = retry.stoptime - time.time()
+            if retry.deadline < 1:
+                raise Etcd3Error('attempt_to_acquire_leader timeout')
+
+        ret = self._run_and_handle_exceptions(self._do_attempt_to_acquire_leader, permanent, retry, retry=None)
         if not ret:
             logger.info('Could not take out TTL lock')
         return ret
@@ -780,17 +809,32 @@ class Etcd3(AbstractEtcd):
         return self._client.put(self.status_path, value)
 
     @catch_etcd_errors
+    def _write_failsafe(self, value):
+        return self._client.put(self.failsafe_path, value)
+
+    @catch_return_false_exception
     def _update_leader(self):
-        if not self._lease:
+        retry = self._retry.copy()
-            self.refresh_lease()
+
-        elif self.retry(self._client.lease_keepalive, self._lease):
+        def _retry(*args, **kwargs):
-            self._last_lease_refresh = time.time()
+            kwargs['retry'] = retry
+            return retry(*args, **kwargs)
+
+        self._run_and_handle_exceptions(self._do_refresh_lease, True, retry=_retry)
 
         if self._lease:
             cluster = self.cluster
             leader_lease = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session
             if leader_lease != self._lease:
-                self.take_leader()
+                retry.deadline = retry.stoptime - time.time()
+                if retry.deadline < 1:
+                    raise Etcd3Error('update_leader timeout')
+
+                try:
+                    self._run_and_handle_exceptions(self._client.put, self.leader_path,
+                                                    self._name, self._lease, retry=_retry)
+                except ReturnFalseException:
+                    pass
         return bool(self._lease)
 
     @catch_etcd_errors

patroni/dcs/kubernetes.py

@@ -1,3 +1,5 @@
+import atexit
+import base64
 import datetime
 import functools
 import json
@@ -7,6 +9,7 @@ import random
 import socket
 import six
 import sys
+import tempfile
 import time
 import urllib3
 import yaml
@@ -28,12 +31,34 @@ SERVICE_HOST_ENV_NAME = 'KUBERNETES_SERVICE_HOST'
 SERVICE_PORT_ENV_NAME = 'KUBERNETES_SERVICE_PORT'
 SERVICE_TOKEN_FILENAME = '/var/run/secrets/kubernetes.io/serviceaccount/token'
 SERVICE_CERT_FILENAME = '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt'
+__temp_files = []
 
 
 class KubernetesError(DCSError):
     pass
 
 
+def _cleanup_temp_files():
+    global __temp_files
+    for temp_file in __temp_files:
+        try:
+            os.remove(temp_file)
+        except OSError:
+            pass
+    __temp_files = []
+
+
+def _create_temp_file(content):
+    if len(__temp_files) == 0:
+        atexit.register(_cleanup_temp_files)
+
+    fd, name = tempfile.mkstemp()
+    os.write(fd, content)
+    os.close(fd)
+    __temp_files.append(name)
+    return name
+
+
 # this function does the same mapping of snake_case => camelCase for > 97% of cases as autogenerated swagger code
 def to_camel_case(value):
     reserved = {'api', 'apiv3', 'cidr', 'cpu', 'csi', 'id', 'io', 'ip', 'ipc', 'pid', 'tls', 'uri', 'url', 'uuid'}
@@ -93,6 +118,13 @@ class K8sConfig(object):
             if c['name'] == name:
                 return c[section]
 
+    def _pool_config_from_file_or_data(self, config, file_key_name, pool_key_name):
+        data_key_name = file_key_name + '-data'
+        if data_key_name in config:
+            self.pool_config[pool_key_name] = _create_temp_file(base64.b64decode(config[data_key_name]))
+        elif file_key_name in config:
+            self.pool_config[pool_key_name] = config[file_key_name]
+
     def load_kube_config(self, context=None):
         with open(os.path.expanduser(KUBE_CONFIG_DEFAULT_LOCATION)) as f:
             config = yaml.safe_load(f)
@@ -103,10 +135,9 @@ class K8sConfig(object):
 
         self._server = cluster['server'].rstrip('/')
         if self._server.startswith('https'):
-            self.pool_config.update({v: user[k] for k, v in {'client-certificate': 'cert_file',
+            self._pool_config_from_file_or_data(user, 'client-certificate', 'cert_file')
-                                                             'client-key': 'key_file'}.items() if k in user})
+            self._pool_config_from_file_or_data(user, 'client-key', 'key_file')
-            if 'certificate-authority' in cluster:
+            self._pool_config_from_file_or_data(cluster, 'certificate-authority', 'ca_certs')
-                self.pool_config['ca_certs'] = cluster['certificate-authority']
             self.pool_config['cert_reqs'] = 'CERT_NONE' if cluster.get('insecure-skip-tls-verify') else 'CERT_REQUIRED'
         if user.get('token'):
             self._make_headers(token=user['token'])
@@ -493,16 +524,10 @@ class CoreV1ApiProxy(object):
 
 
 def catch_kubernetes_errors(func):
-    def wrapper(*args, **kwargs):
+    def wrapper(self, *args, **kwargs):
         try:
-            return func(*args, **kwargs)
+            return self._run_and_handle_exceptions(func, self, *args, **kwargs)
-        except k8s_client.rest.ApiException as e:
+        except KubernetesError:
-            if e.status == 403:
-                logger.exception('Permission denied')
-            elif e.status != 409:  # Object exists or conflict in resource_version
-                logger.exception('Unexpected error from Kubernetes API')
-            return False
-        except (RetryFailedError, K8sException):
             return False
     return wrapper
 
@@ -677,7 +702,7 @@ class Kubernetes(AbstractDCS):
         try:
             k8s_config.load_incluster_config(ca_certs=self._ca_certs)
         except k8s_config.ConfigException:
-            k8s_config.load_kube_config(context=config.get('context', 'local'))
+            k8s_config.load_kube_config(context=config.get('context', 'kind-kind'))
 
         self.__my_pod = None
         self.__ips = [] if config.get('patronictl') else [config.get('pod_ip')]
@@ -712,6 +737,19 @@ class Kubernetes(AbstractDCS):
         kwargs['_retry'] = retry
         return retry(*args, **kwargs)
 
+    @staticmethod
+    def _run_and_handle_exceptions(method, *args, **kwargs):
+        try:
+            return method(*args, **kwargs)
+        except k8s_client.rest.ApiException as e:
+            if e.status == 403:
+                logger.exception('Permission denied')
+            elif e.status != 409:  # Object exists or conflict in resource_version
+                logger.exception('Unexpected error from Kubernetes API')
+            return False
+        except (RetryFailedError, K8sException) as e:
+            raise KubernetesError(e)
+
     def client_path(self, path):
         return super(Kubernetes, self).client_path(path)[1:].replace('/', '-')
 
@@ -794,6 +832,13 @@ class Kubernetes(AbstractDCS):
             except Exception:
                 slots = None
 
+            # get failsafe topology
+            failsafe = annotations.get(self._FAILSAFE)
+            try:
+                failsafe = json.loads(failsafe) if failsafe else None
+            except Exception:
+                failsafe = None
+
             # get leader
             leader_record = {n: annotations.get(n) for n in (self._LEADER, 'acquireTime',
                              'ttl', 'renewTime', 'transitions') if n in annotations}
@@ -826,7 +871,7 @@ class Kubernetes(AbstractDCS):
             metadata = sync and sync.metadata
             sync = SyncState.from_node(metadata and metadata.resource_version,  metadata and metadata.annotations)
 
-            return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
+            return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
         except Exception:
             logger.exception('get_cluster')
             raise KubernetesError('Kubernetes API is not responding properly')
@@ -961,6 +1006,9 @@ class Kubernetes(AbstractDCS):
     def _write_status(self, value):
         """Unused"""
 
+    def _write_failsafe(self, value):
+        """Unused"""
+
     def _update_leader(self):
         """Unused"""
 
@@ -979,16 +1027,19 @@ class Kubernetes(AbstractDCS):
             else:
                 logger.exception('Permission denied' if e.status == 403 else 'Unexpected error from Kubernetes API')
                 return False
-        except (RetryFailedError, K8sException):
+        except (RetryFailedError, K8sException) as e:
-            return False
+            raise KubernetesError(e)
 
+        # if we are here, that means update failed with 409
         retry.deadline = retry.stoptime - time.time()
         if retry.deadline < 1:
-            return False
+            return False  # No time for retry. Tell ha.py that we have to demote due to failed update.
 
         # Try to get the latest version directly from K8s API instead of relying on async cache
         try:
             kind = _retry(self._api.read_namespaced_kind, self.leader_path, self._namespace)
+        except (RetryFailedError, K8sException) as e:
+            raise KubernetesError(e)
         except Exception as e:
             logger.error('Failed to get the leader object "%s": %r', self.leader_path, e)
             return False
@@ -1006,9 +1057,10 @@ class Kubernetes(AbstractDCS):
         if kind and (kind_annotations.get(self._LEADER) != self._name or kind_resource_version == resource_version):
             return False
 
-        return self.patch_or_create(self.leader_path, annotations, kind_resource_version, ips=ips, retry=_retry)
+        return self._run_and_handle_exceptions(self._patch_or_create, self.leader_path, annotations,
+                                               kind_resource_version, ips=ips, retry=_retry)
 
-    def update_leader(self, last_lsn, slots=None):
+    def update_leader(self, last_lsn, slots=None, failsafe=None):
         kind = self._kinds.get(self.leader_path)
         kind_annotations = kind and kind.metadata.annotations or {}
 
@@ -1022,7 +1074,10 @@ class Kubernetes(AbstractDCS):
                        'transitions': leader_observed_record.get('transitions') or '0'}
         if last_lsn:
             annotations[self._OPTIME] = str(last_lsn)
-            annotations['slots'] = json.dumps(slots) if slots else None
+            annotations['slots'] = json.dumps(slots, separators=(',', ':')) if slots else None
+
+        if failsafe is not None:
+            annotations[self._FAILSAFE] = json.dumps(failsafe, separators=(',', ':')) if failsafe else None
 
         resource_version = kind and kind.metadata.resource_version
         return self._update_leader_with_retry(annotations, resource_version, self.__ips)
@@ -1043,7 +1098,19 @@ class Kubernetes(AbstractDCS):
                 annotations['acquireTime'] = self._leader_observed_record.get('acquireTime') or now
             annotations['transitions'] = str(transitions)
         ips = [] if self._api.use_endpoints else None
-        ret = self.patch_or_create(self.leader_path, annotations, self._leader_resource_version, ips=ips)
+
+        try:
+            ret = self._patch_or_create(self.leader_path, annotations,
+                                        self._leader_resource_version, retry=self.retry, ips=ips)
+        except k8s_client.rest.ApiException as e:
+            if e.status == 409 and self._leader_resource_version:  # Conflict in resource_version
+                # Terminate watchers, it could be a sign that K8s API is in a failed state
+                self._kinds.kill_stream()
+                self._pods.kill_stream()
+            ret = False
+        except (RetryFailedError, K8sException) as e:
+            raise KubernetesError(e)
+
         if not ret:
             logger.info('Could not take out TTL lock')
         return ret

patroni/dcs/raft.py

@@ -11,11 +11,16 @@ from pysyncobj.transport import TCPTransport, CONNECTION_STATE
 from pysyncobj.utility import TcpUtility
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
+from ..exceptions import DCSError
 from ..utils import validate_directory
 
 logger = logging.getLogger(__name__)
 
 
+class RaftError(DCSError):
+    pass
+
+
 class _TCPTransport(TCPTransport):
 
     def __init__(self, syncObj, selfNode, otherNodes):
@@ -156,7 +161,7 @@ class KVStoreTTL(DynMemberSyncObj):
             elif deadline:
                 timeout = deadline - time.time()
                 if timeout <= 0:
-                    break
+                    raise RaftError('timeout')
             time.sleep(1)
         return False
 
@@ -175,7 +180,7 @@ class KVStoreTTL(DynMemberSyncObj):
             self.__on_set(key, value)
         return True
 
-    def set(self, key, value, ttl=None, **kwargs):
+    def set(self, key, value, ttl=None, handle_raft_error=True, **kwargs):
         old_value = self.__data.get(key, {})
         if not self.__check_requirements(old_value, **kwargs):
             return False
@@ -184,7 +189,12 @@ class KVStoreTTL(DynMemberSyncObj):
         value['created'] = old_value.get('created', value['updated'])
         if ttl:
             value['expire'] = value['updated'] + ttl
-        return self.retry(self._set, key, value, **kwargs)
+        try:
+            return self.retry(self._set, key, value, **kwargs)
+        except RaftError:
+            if not handle_raft_error:
+                raise
+            return False
 
     def __pop(self, key):
         self.__data.pop(key)
@@ -206,7 +216,10 @@ class KVStoreTTL(DynMemberSyncObj):
     def delete(self, key, recursive=False, **kwargs):
         if not recursive and not self.__check_requirements(self.__data.get(key, {}), **kwargs):
             return False
-        return self.retry(self._delete, key, recursive=recursive, **kwargs)
+        try:
+            return self.retry(self._delete, key, recursive=recursive, **kwargs)
+        except RaftError:
+            return False
 
     @staticmethod
     def __values_match(old, new):
@@ -310,7 +323,7 @@ class Raft(AbstractDCS):
         prefix = self.client_path('')
         response = self._sync_obj.get(prefix, recursive=True)
         if not response:
-            return Cluster(None, None, None, None, [], None, None, None, None)
+            return Cluster(None, None, None, None, [], None, None, None, None, None)
         nodes = {os.path.relpath(key, prefix).replace('\\', '/'): value for key, value in response.items()}
 
         # get initialize flag
@@ -363,7 +376,14 @@ class Raft(AbstractDCS):
         sync = nodes.get(self._SYNC)
         sync = SyncState.from_node(sync and sync['index'], sync and sync['value'])
 
-        return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
+        # get failsafe topology
+        failsafe = nodes.get(self._FAILSAFE)
+        try:
+            failsafe = json.loads(failsafe['value']) if failsafe else None
+        except Exception:
+            failsafe = None
+
+        return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
     def _write_leader_optime(self, last_lsn):
         return self._sync_obj.set(self.leader_optime_path, last_lsn, timeout=1)
@@ -371,15 +391,19 @@ class Raft(AbstractDCS):
     def _write_status(self, value):
         return self._sync_obj.set(self.status_path, value, timeout=1)
 
+    def _write_failsafe(self, value):
+        return self._sync_obj.set(self.failsafe_path, value, timeout=1)
+
     def _update_leader(self):
-        ret = self._sync_obj.set(self.leader_path, self._name, ttl=self._ttl, prevValue=self._name)
+        ret = self._sync_obj.set(self.leader_path, self._name, ttl=self._ttl,
+                                 handle_raft_error=False, prevValue=self._name)
         if not ret and self._sync_obj.get(self.leader_path) is None:
             ret = self.attempt_to_acquire_leader()
         return ret
 
     def attempt_to_acquire_leader(self, permanent=False):
-        return self._sync_obj.set(self.leader_path, self._name, prevExist=False,
+        return self._sync_obj.set(self.leader_path, self._name, ttl=None if permanent else self._ttl,
-                                  ttl=None if permanent else self._ttl)
+                                  handle_raft_error=False, prevExist=False)
 
     def set_failover_value(self, value, index=None):
         return self._sync_obj.set(self.failover_path, value, prevIndex=index)

patroni/dcs/zookeeper.py

@@ -5,9 +5,10 @@ import six
 import time
 
 from kazoo.client import KazooClient, KazooState, KazooRetry
-from kazoo.exceptions import NoNodeError, NodeExistsError, SessionExpiredError
+from kazoo.exceptions import ConnectionClosedError, NoNodeError, NodeExistsError, SessionExpiredError
 from kazoo.handlers.threading import SequentialThreadingHandler
 from kazoo.protocol.states import KeeperState
+from kazoo.retry import RetryFailedError
 from kazoo.security import make_acl
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
@@ -263,18 +264,10 @@ class ZooKeeper(AbstractDCS):
         # get leader
         leader = self.get_node(self.leader_path) if self._LEADER in nodes else None
         if leader:
-            client_id = self._client.client_id
+            member = Member(-1, leader[0], None, {})
-            if not self._ctl and leader[0] == self._name and client_id is not None \
+            member = ([m for m in members if m.name == leader[0]] or [member])[0]
-                    and client_id[0] != leader[1].ephemeralOwner:
+            leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
-                logger.info('I am leader but not owner of the session. Removing leader node')
+            self._fetch_cluster = member.index == -1
-                self._client.delete(self.leader_path)
-                leader = None
-
-            if leader:
-                member = Member(-1, leader[0], None, {})
-                member = ([m for m in members if m.name == leader[0]] or [member])[0]
-                leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
-                self._fetch_cluster = member.index == -1
 
         # get last known leader lsn and slots
         last_lsn, slots = self.get_status(leader)
@@ -283,7 +276,14 @@ class ZooKeeper(AbstractDCS):
         failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
         failover = failover and Failover.from_node(failover[1].version, failover[0])
 
-        return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots)
+        # get failsafe topology
+        failsafe = self.get_node(self.failsafe_path, watch=self.cluster_watcher) if self._FAILSAFE in nodes else None
+        try:
+            failsafe = json.loads(failsafe[0]) if failsafe else None
+        except Exception:
+            failsafe = None
+
+        return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
 
     def _load_cluster(self):
         cluster = self.cluster
@@ -304,8 +304,8 @@ class ZooKeeper(AbstractDCS):
                 try:
                     last_lsn, slots = self.get_status(cluster.leader)
                     self.event.clear()
-                    cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, last_lsn,
+                    cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, last_lsn, cluster.members,
-                                      cluster.members, cluster.failover, cluster.sync, cluster.history, slots)
+                                      cluster.failover, cluster.sync, cluster.history, slots, cluster.failsafe)
                 except Exception:
                     pass
         return cluster
@@ -325,10 +325,17 @@ class ZooKeeper(AbstractDCS):
         return False
 
     def attempt_to_acquire_leader(self, permanent=False):
-        ret = self._create(self.leader_path, self._name.encode('utf-8'), retry=True, ephemeral=not permanent)
+        try:
-        if not ret:
+            self._client.retry(self._client.create, self.leader_path, self._name.encode('utf-8'),
-            logger.info('Could not take out TTL lock')
+                               makepath=True, ephemeral=not permanent)
-        return ret
+            return True
+        except (ConnectionClosedError, RetryFailedError) as e:
+            raise ZooKeeperError(e)
+        except Exception as e:
+            if not isinstance(e, NodeExistsError):
+                logger.error('Failed to create %s: %r', self.leader_path, e)
+        logger.info('Could not take out TTL lock')
+        return False
 
     def _set_or_create(self, key, value, index=None, retry=False, do_not_create_empty=False):
         value = value.encode('utf-8')
@@ -408,7 +415,32 @@ class ZooKeeper(AbstractDCS):
     def _write_status(self, value):
         return self._set_or_create(self.status_path, value)
 
+    def _write_failsafe(self, value):
+        return self._set_or_create(self.failsafe_path, value)
+
     def _update_leader(self):
+        cluster = self.cluster
+        session = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session
+        if self._client.client_id and self._client.client_id[0] != session:
+            logger.warning('Recreating the leader ZNode due to ownership mismatch')
+            try:
+                self._client.retry(self._client.delete, self.leader_path)
+            except NoNodeError:
+                pass
+            except (ConnectionClosedError, RetryFailedError) as e:
+                raise ZooKeeperError(e)
+            except Exception as e:
+                logger.error('Failed to remove %s: %r', self.leader_path, e)
+                return False
+
+            try:
+                self._client.retry(self._client.create, self.leader_path,
+                                   self._name.encode('utf-8'), makepath=True, ephemeral=True)
+            except (ConnectionClosedError, RetryFailedError) as e:
+                raise ZooKeeperError(e)
+            except Exception as e:
+                logger.error('Failed to create %s: %r', self.leader_path, e)
+                return False
         return True
 
     def _delete_leader(self):

patroni/ha.py

@@ -39,11 +39,20 @@ class _MemberStatus(namedtuple('_MemberStatus', ['member', 'reachable', 'in_reco
     """
     @classmethod
     def from_api_response(cls, member, json):
-        is_master = json['role'] == 'master'
+        """
+        :param member: dcs.Member object
+        :param json: RestApiHandler.get_postgresql_status() result
+        :returns: _MemberStatus object
+        """
+        # If one of those is not in a response we want to count the node as not healthy/reachable
+        assert 'wal' in json or 'xlog' in json
+
+        wal = json.get('wal', json.get('xlog'))
+        in_recovery = not bool(wal.get('location'))  # abuse difference in primary/replica response format
         timeline = json.get('timeline', 0)
         dcs_last_seen = json.get('dcs_last_seen', 0)
-        wal = not is_master and max(json['xlog'].get('received_location', 0), json['xlog'].get('replayed_location', 0))
+        wal = in_recovery and max(wal.get('received_location', 0), wal.get('replayed_location', 0))
-        return cls(member, True, not is_master, dcs_last_seen, timeline, wal,
+        return cls(member, True, in_recovery, dcs_last_seen, timeline, wal,
                    json.get('tags', {}), json.get('watchdog_failed', False))
 
     @classmethod
@@ -146,7 +155,13 @@ class Ha(object):
         self._leader_timeline = None if cluster.is_unlocked() else cluster.leader.timeline
 
     def acquire_lock(self):
-        ret = self.dcs.attempt_to_acquire_leader()
+        try:
+            ret = self.dcs.attempt_to_acquire_leader()
+        except DCSError:
+            raise
+        except Exception:
+            logger.exception('Unexpected exception raised from attempt_to_acquire_leader, please report it as a BUG')
+            ret = False
         self.set_is_leader(ret)
         return ret
 
@@ -160,6 +175,8 @@ class Ha(object):
                 logger.exception('Exception when called state_handler.last_operation()')
         try:
             ret = self.dcs.update_leader(last_lsn, slots)
+        except DCSError:
+            raise
         except Exception:
             logger.exception('Unexpected exception raised from update_leader, please report it as a BUG')
             ret = False
@@ -1522,8 +1539,12 @@ class Ha(object):
             dcs_failed = True
             logger.error('Error communicating with DCS')
             if not self.is_paused() and self.state_handler.is_running() and self.state_handler.is_leader():
+                msg = 'demoting self because DCS is not accessible and I was a leader'
+                if not self._async_executor.try_run_async(msg, self.demote, ('offline',)):
+                    return msg
+                logger.warning('AsyncExecutor is busy, demoting from the main thread')
                 self.demote('offline')
-                return 'demoted self because DCS is not accessible and i was a leader'
+                return 'demoted self because DCS is not accessible and I was a leader'
             return 'DCS is not accessible'
         except (psycopg.Error, PostgresConnectionException):
             return 'Error communicating with PostgreSQL. Will try again later'

patroni/postgresql/bootstrap.py

@@ -315,12 +315,14 @@ END;$$""".format(quote_literal(name), quote_ident(name, self._postgresql.connect
         self._postgresql.query('SET log_statement TO none')
         self._postgresql.query('SET log_min_duration_statement TO -1')
         self._postgresql.query("SET log_min_error_statement TO 'log'")
+        self._postgresql.query("SET pg_stat_statements.track_utility to 'off'")
         try:
             self._postgresql.query(sql)
         finally:
             self._postgresql.query('RESET log_min_error_statement')
             self._postgresql.query('RESET log_min_duration_statement')
             self._postgresql.query('RESET log_statement')
+            self._postgresql.query('RESET pg_stat_statements.track_utility')
 
     def post_bootstrap(self, config, task):
         try:

patroni/postgresql/connection.py

@@ -22,7 +22,6 @@ class Connection(object):
         with self._lock:
             if not self._connection or self._connection.closed != 0:
                 self._connection = psycopg.connect(**self._conn_kwargs)
-                self._connection.autocommit = True
                 self.server_version = self._connection.server_version
         return self._connection
 
@@ -42,7 +41,6 @@ class Connection(object):
 @contextmanager
 def get_connection_cursor(**kwargs):
     conn = psycopg.connect(**kwargs)
-    conn.autocommit = True
     with conn.cursor() as cur:
         yield cur
     conn.close()

patroni/psycopg.py

@@ -6,7 +6,7 @@ try:
     from . import MIN_PSYCOPG2, parse_version
     if parse_version(__version__) < MIN_PSYCOPG2:
         raise ImportError
-    from psycopg2 import connect, Error, DatabaseError, OperationalError, ProgrammingError
+    from psycopg2 import connect as _connect, Error, DatabaseError, OperationalError, ProgrammingError
     from psycopg2.extensions import adapt
 
     try:
@@ -20,10 +20,10 @@ try:
             value.prepare(conn)
         return value.getquoted().decode('utf-8')
 except ImportError:
-    from psycopg import connect as _connect, sql, Error, DatabaseError, OperationalError, ProgrammingError
+    from psycopg import connect as __connect, sql, Error, DatabaseError, OperationalError, ProgrammingError
 
-    def connect(*args, **kwargs):
+    def _connect(*args, **kwargs):
-        ret = _connect(*args, **kwargs)
+        ret = __connect(*args, **kwargs)
         ret.server_version = ret.pgconn.server_version  # compatibility with psycopg2
         return ret
 
@@ -34,6 +34,16 @@ except ImportError:
         return sql.Literal(value).as_string(conn)
 
 
+def connect(*args, **kwargs):
+    if kwargs and 'replication' not in kwargs and kwargs.get('fallback_application_name') != 'Patroni ctl':
+        options = [kwargs['options']] if 'options' in kwargs else []
+        options.append('-c search_path=pg_catalog')
+        kwargs['options'] = ' '.join(options)
+    ret = _connect(*args, **kwargs)
+    ret.autocommit = True
+    return ret
+
+
 def quote_ident(value, conn=None):
     if _legacy or conn is None:
         return '"{0}"'.format(value.replace('"', '""'))

patroni/request.py

@@ -9,9 +9,9 @@ from .utils import USER_AGENT
 
 class PatroniRequest(object):
 
-    def __init__(self, config, insecure=False):
+    def __init__(self, config, insecure=None):
-        cert_reqs = 'CERT_NONE' if insecure or config.get('ctl', {}).get('insecure', False) else 'CERT_REQUIRED'
+        self._insecure = insecure
-        self._pool = urllib3.PoolManager(num_pools=10, maxsize=10, cert_reqs=cert_reqs)
+        self._pool = urllib3.PoolManager(num_pools=10, maxsize=10)
         self.reload_config(config)
 
     @staticmethod
@@ -32,12 +32,19 @@ class PatroniRequest(object):
     def reload_config(self, config):
         self._pool.headers = urllib3.make_headers(basic_auth=self._get_cfg_value(config, 'auth'), user_agent=USER_AGENT)
 
+        insecure = self._insecure if isinstance(self._insecure, bool) else config.get('ctl', {}).get('insecure', False)
         if self._apply_ssl_file_param(config, 'cert'):
+            #  With client certificate the cert_reqs must be set to CERT_REQUIRED even if insecure option is used
+            self._pool.connection_pool_kw['cert_reqs'] = 'CERT_REQUIRED'
+            #  The assert_hostname = False helps to silence warnings
+            self._pool.connection_pool_kw['assert_hostname'] = False if insecure else None
+
             self._apply_ssl_file_param(config, 'key')
 
             password = self._get_cfg_value(config, 'keyfile_password')
             self._apply_pool_param('key_password', password)
         else:
+            self._pool.connection_pool_kw['cert_reqs'] = 'CERT_NONE' if insecure else 'CERT_REQUIRED'
             self._pool.connection_pool_kw.pop('key_file', None)
 
         cacert = config.get('ctl', {}).get('cacert') or config.get('restapi', {}).get('cafile')

patroni/scripts/wale_restore.py

@@ -214,26 +214,26 @@ class WALERestore(object):
         attempts_no = 0
         while True:
             if self.master_connection:
+                con = None
                 try:
                     # get the difference in bytes between the current WAL location and the backup start offset
-                    with psycopg.connect(self.master_connection) as con:
+                    con = psycopg.connect(self.master_connection)
-                        if con.server_version >= 100000:
+                    if con.server_version >= 100000:
-                            wal_name = 'wal'
+                        wal_name = 'wal'
-                            lsn_name = 'lsn'
+                        lsn_name = 'lsn'
-                        else:
+                    else:
-                            wal_name = 'xlog'
+                        wal_name = 'xlog'
-                            lsn_name = 'location'
+                        lsn_name = 'location'
-                        con.autocommit = True
+                    with con.cursor() as cur:
-                        with con.cursor() as cur:
+                        cur.execute(("SELECT CASE WHEN pg_catalog.pg_is_in_recovery()"
-                            cur.execute(("SELECT CASE WHEN pg_catalog.pg_is_in_recovery()"
+                                     " THEN GREATEST(pg_catalog.pg_{0}_{1}_diff(COALESCE("
-                                         " THEN GREATEST(pg_catalog.pg_{0}_{1}_diff(COALESCE("
+                                     "pg_last_{0}_receive_{1}(), '0/0'), %s)::bigint, "
-                                         "pg_last_{0}_receive_{1}(), '0/0'), %s)::bigint, "
+                                     "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), %s)::bigint)"
-                                         "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), %s)::bigint)"
+                                     " ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), %s)::bigint"
-                                         " ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), %s)::bigint"
+                                     " END").format(wal_name, lsn_name),
-                                         " END").format(wal_name, lsn_name),
+                                    (backup_start_lsn, backup_start_lsn, backup_start_lsn))
-                                        (backup_start_lsn, backup_start_lsn, backup_start_lsn))
 
-                            diff_in_bytes = int(cur.fetchone()[0])
+                        diff_in_bytes = int(cur.fetchone()[0])
                 except psycopg.Error:
                     logger.exception('could not determine difference with the master location')
                     if attempts_no < self.retries:  # retry in case of a temporarily connection issue
@@ -246,6 +246,9 @@ class WALERestore(object):
                         logger.info("continue with base backup from S3 since master is not available")
                         diff_in_bytes = 0
                         break
+                finally:
+                    if con:
+                        con.close()
             else:
                 # always try to use WAL-E if master connection string is not available
                 diff_in_bytes = 0

patroni/utils.py

@@ -519,8 +519,8 @@ def enable_keepalive(sock, timeout, idle, cnt=3):
 def find_executable(executable, path=None):
     _, ext = os.path.splitext(executable)
 
-    if (sys.platform == 'win32') and (ext != '.exe'):
+    if (sys.platform == 'win32') and (ext == ''):
-        executable = executable + '.exe'
+        executable = executable + '.exe'  # Set default WIN extension
 
     if os.path.isfile(executable):
         return executable

patroni/validator.py

@@ -370,7 +370,7 @@ schema = Schema({
   "postgresql": {
     "listen": validate_host_port_listen_multiple_hosts,
     "connect_address": validate_connect_address,
-    "proxy_address": validate_connect_address,
+    Optional("proxy_address"): validate_connect_address,
     "authentication": {
       "replication": userattributes,
       "superuser": userattributes,

patroni/version.py

@@ -1 +1 @@
-__version__ = '2.1.5'
+__version__ = '2.1.6'

patroni/watchdog/base.py

@@ -215,6 +215,10 @@ class Watchdog(object):
                     self._activate()
                 if self.config.timeout != self.active_config.timeout:
                     self.impl.set_timeout(self.config.timeout)
+                    if self.is_running:
+                        logger.info("{0} updated with {1} second timeout, timing slack {2} seconds"
+                                    .format(self.impl.describe(), self.impl.get_timeout(), self.config.timing_slack))
+                self.active_config = self.config
         except WatchdogError as e:
             logger.error("Error while sending keepalive: %s", e)
 

postgres0.yml

@@ -5,16 +5,18 @@ name: postgresql0
 restapi:
   listen: 127.0.0.1:8008
   connect_address: 127.0.0.1:8008
+#  cafile: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 #  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
 #  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
 #  authentication:
 #    username: username
 #    password: password
 
-# ctl:
+#ctl:
-#   insecure: false # Allow connections to SSL sites without certs
+#  insecure: false # Allow connections to Patroni REST API without verifying certificates
-#   certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
+#  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
-#   cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
+#  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
+#  cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 
 etcd:
   #Provide host to do the initial discovery of the cluster topology:

postgres1.yml

@@ -5,16 +5,18 @@ name: postgresql1
 restapi:
   listen: 127.0.0.1:8009
   connect_address: 127.0.0.1:8009
+#  cafile: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 #  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
 #  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
 #  authentication:
 #    username: username
 #    password: password
 
-# ctl:
+#ctl:
-#   insecure: false # Allow connections to SSL sites without certs
+#  insecure: false # Allow connections to Patroni REST API without verifying certificates
-#   certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
+#  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
-#   cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
+#  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
+#  cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 
 etcd:
   #Provide host to do the initial discovery of the cluster topology:

postgres2.yml

@@ -5,16 +5,18 @@ name: postgresql2
 restapi:
   listen: 127.0.0.1:8010
   connect_address: 127.0.0.1:8010
+#  cafile: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 #  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
 #  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
   authentication:
     username: username
     password: password
 
-# ctl:
+#ctl:
-#   insecure: false # Allow connections to SSL sites without certs
+#  insecure: false # Allow connections to Patroni REST API without verifying certificates
-#   certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
+#  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
-#   cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
+#  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
+#  cacert: /etc/ssl/certs/ssl-cacert-snakeoil.pem
 
 etcd:
   #Provide host to do the initial discovery of the cluster topology:

tests/__init__.py

@@ -50,7 +50,7 @@ def requests_get(url, **kwargs):
     if url.startswith('http://local'):
         raise urllib3.exceptions.HTTPError()
     elif ':8011/patroni' in url:
-        response.content = '{"role": "replica", "xlog": {"received_location": 0}, "tags": {}}'
+        response.content = '{"role": "replica", "wal": {"received_location": 0}, "tags": {}}'
     elif url.endswith('/members'):
         response.content = '[{}]' if url.startswith('http://error') else members
     elif url.startswith('http://exhibitor'):
@@ -177,7 +177,7 @@ class PostgresInit(unittest.TestCase):
                    'force_parallel_mode': '1', 'constraint_exclusion': '',
                    'max_stack_depth': 'Z', 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200}
 
-    @patch('patroni.psycopg.connect', psycopg_connect)
+    @patch('patroni.psycopg._connect', psycopg_connect)
     @patch('patroni.postgresql.CallbackExecutor', Mock())
     @patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
     @patch.object(ConfigHandler, 'replace_pg_hba', Mock())

tests/test_api.py

@@ -601,8 +601,11 @@ class TestRestApiServer(unittest.TestCase):
         self.srv.process_request_thread(Mock(), '2')
 
     @patch.object(MockRestApiServer, 'process_request', Mock(side_effect=RuntimeError))
-    @patch.object(MockRestApiServer, 'get_request', Mock(return_value=(Mock(), ('127.0.0.1', 55555))))
+    @patch.object(MockRestApiServer, 'get_request')
-    def test_process_request_error(self):
+    def test_process_request_error(self, mock_get_request):
+        mock_request = Mock()
+        mock_request.unwrap.side_effect = Exception
+        mock_get_request.return_value = (mock_request, ('127.0.0.1', 55555))
         self.srv._handle_request_noblock()
 
     @patch('ssl._ssl._test_decode_cert', Mock())

tests/test_consul.py

@@ -4,7 +4,7 @@ import unittest
 from consul import ConsulException, NotFound
 from mock import Mock, PropertyMock, patch
 from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \
-                                ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession
+        ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError
 from . import SleepException
 
 
@@ -34,6 +34,8 @@ def kv_get(self, key, **kwargs):
                   'ModifyIndex': 6429, 'Value': b'4496294792'},
                  {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'sync', 'LockIndex': 0,
                   'ModifyIndex': 6429, 'Value': b'{"leader": "leader", "sync_standby": null}'},
+                 {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'failsafe', 'LockIndex': 0,
+                  'ModifyIndex': 6429, 'Value': b'{'},
                  {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'status', 'LockIndex': 0,
                   'ModifyIndex': 6429, 'Value': b'{"optime":4496294792, "slots":{"ls":12345}}'}])
     if key == 'service/good/':
@@ -122,9 +124,6 @@ class TestConsul(unittest.TestCase):
         self.assertIsInstance(self.c.get_cluster(), Cluster)
         self.c._base_path = '/service/legacy'
         self.assertIsInstance(self.c.get_cluster(), Cluster)
-        self.c._base_path = '/service/good'
-        self.c._session = 'fd4f44fe-2cac-bba5-a60b-304b51ff39b8'
-        self.assertIsInstance(self.c.get_cluster(), Cluster)
 
     @patch.object(consul.Consul.KV, 'delete', Mock(side_effect=[ConsulException, True, True, True]))
     @patch.object(consul.Consul.KV, 'put', Mock(side_effect=[True, ConsulException, InvalidSession]))
@@ -140,11 +139,15 @@ class TestConsul(unittest.TestCase):
         self.c.refresh_session = Mock(side_effect=ConsulError('foo'))
         self.assertFalse(self.c.touch_member({'balbla': 'blabla'}))
 
-    @patch.object(consul.Consul.KV, 'put', Mock(side_effect=InvalidSession))
+    @patch.object(consul.Consul.KV, 'put', Mock(side_effect=[InvalidSession, False, InvalidSession]))
     def test_take_leader(self):
         self.c.set_ttl(20)
-        self.c.refresh_session = Mock()
+        self.c._do_refresh_session = Mock()
-        self.c.take_leader()
+        self.assertFalse(self.c.take_leader())
+        with patch('time.time', Mock(side_effect=[0, 100])):
+            self.assertRaises(ConsulError, self.c.take_leader)
+        with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 0, 100])):
+            self.assertRaises(ConsulError, self.c.take_leader)
 
     @patch.object(consul.Consul.KV, 'put', Mock(return_value=True))
     def test_set_failover_value(self):
@@ -160,10 +163,26 @@ class TestConsul(unittest.TestCase):
         self.c.get_cluster()
         self.c.write_leader_optime('1')
 
-    @patch.object(consul.Consul.Session, 'renew', Mock())
+    @patch.object(consul.Consul.Session, 'renew')
     @patch.object(consul.Consul.KV, 'put', Mock(side_effect=ConsulException))
-    def test_update_leader(self):
+    def test_update_leader(self, mock_renew):
-        self.c.update_leader(12345)
+        self.c._session = 'fd4f44fe-2cac-bba5-a60b-304b51ff39b8'
+        with patch.object(consul.Consul.KV, 'delete', Mock(return_value=True)):
+            with patch.object(consul.Consul.KV, 'put', Mock(return_value=True)):
+                self.assertTrue(self.c.update_leader(12345, failsafe={'foo': 'bar'}))
+            with patch.object(consul.Consul.KV, 'put', Mock(side_effect=ConsulException)):
+                self.assertFalse(self.c.update_leader(12345))
+            with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 100, 200, 300])):
+                self.assertRaises(ConsulError, self.c.update_leader, 12345)
+        with patch('time.time', Mock(side_effect=[0, 100, 200, 300])):
+            self.assertRaises(ConsulError, self.c.update_leader, 12345)
+        with patch.object(consul.Consul.KV, 'delete', Mock(side_effect=ConsulException)):
+            self.assertFalse(self.c.update_leader(12347))
+        mock_renew.side_effect = RetryFailedError('')
+        self.c._last_session_refresh = 0
+        self.assertRaises(ConsulError, self.c.update_leader, 12346)
+        mock_renew.side_effect = ConsulException
+        self.assertFalse(self.c.update_leader(12347))
 
     @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True))
     def test_delete_leader(self):

tests/test_ctl.py

@@ -5,8 +5,8 @@ import unittest
 from click.testing import CliRunner
 from datetime import datetime, timedelta
 from mock import patch, Mock
-from patroni.ctl import ctl, store_config, load_config, output_members, get_dcs, parse_dcs, \
+from patroni.ctl import ctl, load_config, output_members, get_dcs, parse_dcs, \
-    get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException, apply_config_changes, \
+    get_all_members, get_any_member, get_cursor, query_member, PatroniCtlException, apply_config_changes, \
     format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH, PatronictlPrettyTable
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover, Failover
 from patroni.psycopg import OperationalError
@@ -20,17 +20,6 @@ from .test_ha import get_cluster_initialized_without_leader, get_cluster_initial
     get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader, get_cluster, Member
 
 
-def test_rw_config():
-    runner = CliRunner()
-    with runner.isolated_filesystem():
-        load_config(CONFIG_FILE_PATH, None)
-        CONFIG_PATH = './test-ctl.yaml'
-        store_config({'etcd': {'host': 'localhost:2379'}}, CONFIG_PATH + '/dummy')
-        load_config(CONFIG_PATH + '/dummy', '0.0.0.0')
-        os.remove(CONFIG_PATH + '/dummy')
-        os.rmdir(CONFIG_PATH)
-
-
 @patch('patroni.ctl.load_config', Mock(return_value={
     'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'}, 'etcd': {'host': 'localhost:2379'},
     'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5}}))
@@ -43,11 +32,27 @@ class TestCtl(unittest.TestCase):
             self.runner = CliRunner()
             self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}}, 'foo')
 
-    def test_load_config(self):
+    @patch('patroni.ctl.logging.debug')
+    def test_load_config(self, mock_logger_debug):
         runner = CliRunner()
         with runner.isolated_filesystem():
             self.assertRaises(PatroniCtlException, load_config, './non-existing-config-file', None)
-            self.assertRaises(PatroniCtlException, load_config, './non-existing-config-file', None)
+
+        with patch('os.path.exists', Mock(return_value=True)), \
+             patch('patroni.config.Config._load_config_path', Mock(return_value={})):
+            load_config(CONFIG_FILE_PATH, None)
+            mock_logger_debug.assert_called_once()
+            self.assertEqual(('Ignoring configuration file "%s". It does not exists or is not readable.',
+                              CONFIG_FILE_PATH),
+                             mock_logger_debug.call_args[0])
+            mock_logger_debug.reset_mock()
+
+            with patch('os.access', Mock(return_value=True)):
+                load_config(CONFIG_FILE_PATH, '')
+                mock_logger_debug.assert_called_once()
+                self.assertEqual(('Loading configuration from file %s', CONFIG_FILE_PATH),
+                                 mock_logger_debug.call_args[0])
+                mock_logger_debug.reset_mock()
 
     @patch('patroni.psycopg.connect', psycopg_connect)
     def test_get_cursor(self):
@@ -380,10 +385,6 @@ class TestCtl(unittest.TestCase):
         with patch('patroni.ctl.load_config', Mock(return_value={})):
             self.runner.invoke(ctl, ['list'])
 
-    def test_configure(self):
-        result = self.runner.invoke(configure, ['--dcs', 'abc', '-c', 'dummy', '-n', 'bla'])
-        assert result.exit_code == 0
-
     @patch('patroni.ctl.get_dcs')
     def test_scaffold(self, mock_get_dcs):
         mock_get_dcs.return_value = self.e

tests/test_etcd.py

@@ -67,6 +67,7 @@ def etcd_read(self, key, **kwargs):
                      "expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30,
                      "modifiedIndex": 20730, "createdIndex": 20730}],
                  "modifiedIndex": 1581, "createdIndex": 1581},
+                {"key": "/service/batman5/failsafe", "value": '{', "modifiedIndex": 1582, "createdIndex": 1582},
                 {"key": "/service/batman5/status", "value": '{"optime":2164261704,"slots":{"ls":12345}}',
                  "modifiedIndex": 1582, "createdIndex": 1582}], "modifiedIndex": 1581, "createdIndex": 1581}}
     if key == '/service/legacy/':
@@ -276,6 +277,9 @@ class TestEtcd(unittest.TestCase):
         self.assertFalse(self.etcd.attempt_to_acquire_leader())
         self.etcd._base_path = '/service/failed'
         self.assertFalse(self.etcd.attempt_to_acquire_leader())
+        with patch.object(EtcdClient, 'write', Mock(side_effect=[etcd.EtcdConnectionFailed, Exception])):
+            self.assertRaises(EtcdError, self.etcd.attempt_to_acquire_leader)
+            self.assertRaises(EtcdError, self.etcd.attempt_to_acquire_leader)
 
     @patch.object(Cluster, 'min_version', PropertyMock(return_value=(2, 0)))
     def test_write_leader_optime(self):
@@ -283,7 +287,14 @@ class TestEtcd(unittest.TestCase):
         self.etcd.write_leader_optime('0')
 
     def test_update_leader(self):
-        self.assertTrue(self.etcd.update_leader(None))
+        self.assertTrue(self.etcd.update_leader(None, failsafe={'foo': 'bar'}))
+        with patch.object(etcd.Client, 'write',
+                          Mock(side_effect=[etcd.EtcdConnectionFailed, etcd.EtcdClusterIdChanged, Exception])):
+            self.assertRaises(EtcdError, self.etcd.update_leader, None)
+            self.assertFalse(self.etcd.update_leader(None))
+            self.assertRaises(EtcdError, self.etcd.update_leader, None)
+        with patch.object(etcd.Client, 'write', Mock(side_effect=etcd.EtcdKeyNotFound)):
+            self.assertFalse(self.etcd.update_leader(None))
 
     def test_initialize(self):
         self.assertFalse(self.etcd.initialize())

tests/test_etcd3.py

@@ -36,7 +36,8 @@ def mock_urlopen(self, method, url, **kwargs):
                  "value": base64_encode('{}'), "lease": "123", "mod_revision": '1'},
                 {"key": base64_encode('/patroni/test/members/bar'),
                  "value": base64_encode('{"version":"1.6.5"}'), "lease": "123", "mod_revision": '1'},
-                {"key": base64_encode('/patroni/test/failover'), "value": base64_encode('{}'), "mod_revision": '1'}
+                {"key": base64_encode('/patroni/test/failover'), "value": base64_encode('{}'), "mod_revision": '1'},
+                {"key": base64_encode('/patroni/test/failsafe'), "value": base64_encode('{'), "mod_revision": '1'}
             ]
         })
     elif url.endswith('/watch'):
@@ -215,12 +216,27 @@ class TestEtcd3(BaseTestEtcd3):
 
     def test__update_leader(self):
         self.etcd3._lease = None
-        self.etcd3.update_leader('123')
+        self.etcd3.update_leader('123', failsafe={'foo': 'bar'})
+        self.etcd3._last_lease_refresh = 0
         self.etcd3.update_leader('124')
+        with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)),\
+                patch('time.time', Mock(side_effect=[0, 100, 200, 300])):
+            self.assertRaises(Etcd3Error, self.etcd3.update_leader, '126')
+        self.etcd3._last_lease_refresh = 0
+        with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(side_effect=Unknown)):
+            self.assertFalse(self.etcd3.update_leader('125'))
+
+    def test_take_leader(self):
+        self.assertFalse(self.etcd3.take_leader())
 
     def test_attempt_to_acquire_leader(self):
-        self.etcd3._lease = None
         self.assertFalse(self.etcd3.attempt_to_acquire_leader())
+        with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 100, 200])):
+            self.assertRaises(Etcd3Error, self.etcd3.attempt_to_acquire_leader)
+        with patch('time.time', Mock(side_effect=[0, 100, 200, 300, 400])):
+            self.assertRaises(Etcd3Error, self.etcd3.attempt_to_acquire_leader)
+        with patch.object(PatroniEtcd3Client, 'put', Mock(return_value=False)):
+            self.assertFalse(self.etcd3.attempt_to_acquire_leader())
 
     def test_set_ttl(self):
         self.etcd3.set_ttl(20)

tests/test_exhibitor.py

@@ -30,5 +30,6 @@ class TestExhibitor(unittest.TestCase):
                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
 
     @patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True))
+    @patch.object(MockKazooClient, 'get_children', Mock(side_effect=Exception))
     def test_get_cluster(self):
         self.assertRaises(ZooKeeperError, self.e.get_cluster)

tests/test_ha.py

@@ -38,7 +38,7 @@ def get_cluster(initialize, leader, members, failover, sync, cluster_config=None
     history = TimelineHistory(1, '[[1,67197376,"no recovery target specified","' + t + '","foo"]]',
                               [(1, 67197376, 'no recovery target specified', t, 'foo')])
     cluster_config = cluster_config or ClusterConfig(1, {'check_timeline': True}, 1)
-    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history, None)
+    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history, None, None)
 
 
 def get_cluster_not_initialized_without_leader(cluster_config=None):
@@ -206,7 +206,8 @@ class TestHa(PostgresInit):
 
     def test_update_lock(self):
         self.p.last_operation = Mock(side_effect=PostgresConnectionException(''))
-        self.ha.dcs.update_leader = Mock(side_effect=Exception)
+        self.ha.dcs.update_leader = Mock(side_effect=[DCSError(''), Exception])
+        self.assertRaises(DCSError, self.ha.update_lock)
         self.assertFalse(self.ha.update_lock(True))
 
     @patch.object(Postgresql, 'received_timeline', Mock(return_value=None))
@@ -458,7 +459,9 @@ class TestHa(PostgresInit):
 
     def test_no_etcd_connection_master_demote(self):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
-        self.assertEqual(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader')
+        self.assertEqual(self.ha.run_cycle(), 'demoting self because DCS is not accessible and I was a leader')
+        self.ha._async_executor.schedule('dummy')
+        self.assertEqual(self.ha.run_cycle(), 'demoted self because DCS is not accessible and I was a leader')
 
     @patch('time.sleep', Mock())
     def test_bootstrap_from_another_member(self):
@@ -1285,3 +1288,8 @@ class TestHa(PostgresInit):
         self.ha.fetch_node_status = Mock(return_value=_MemberStatus(self.ha.cluster.members[0],
                                                                     True, True, 0, 2, None, {}, False))
         self.assertFalse(self.ha.is_failover_possible(self.ha.cluster.members))
+
+    def test_acquire_lock(self):
+        self.ha.dcs.attempt_to_acquire_leader = Mock(side_effect=[DCSError('foo'), Exception])
+        self.assertRaises(DCSError, self.ha.acquire_lock)
+        self.assertFalse(self.ha.acquire_lock())

tests/test_kubernetes.py

@@ -1,10 +1,11 @@
+import base64
 import datetime
 import json
 import socket
 import time
 import unittest
 
-from mock import Mock, PropertyMock, mock_open, patch
+from mock import call, Mock, PropertyMock, mock_open, patch
 from patroni.dcs.kubernetes import k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\
         K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\
         Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
@@ -18,7 +19,7 @@ def mock_list_namespaced_config_map(*args, **kwargs):
                 'annotations': {'initialize': '123', 'config': '{}'}}
     items = [k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))]
     metadata.update({'name': 'test-leader',
-                     'annotations': {'optime': '1234x', 'leader': 'p-0', 'ttl': '30s', 'slots': '{'}})
+                     'annotations': {'optime': '1234x', 'leader': 'p-0', 'ttl': '30s', 'slots': '{', 'failsafe': '{'}})
     items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
     metadata.update({'name': 'test-failover', 'annotations': {'leader': 'p-0'}})
     items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata)))
@@ -121,6 +122,20 @@ class TestK8sConfig(unittest.TestCase):
             k8s_config.load_kube_config()
             self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer token')
 
+        config["users"][0]["user"]["client-key-data"] = base64.b64encode(b'foobar').decode('utf-8')
+        config["clusters"][0]["cluster"]["certificate-authority-data"] = base64.b64encode(b'foobar').decode('utf-8')
+        with patch.object(builtins, 'open', mock_open(read_data=json.dumps(config))),\
+                patch('os.write', Mock()), patch('os.close', Mock()),\
+                patch('os.remove') as mock_remove,\
+                patch('atexit.register') as mock_atexit,\
+                patch('tempfile.mkstemp') as mock_mkstemp:
+            mock_mkstemp.side_effect = [(3, '1.tmp'), (4, '2.tmp')]
+            k8s_config.load_kube_config()
+            mock_atexit.assert_called_once()
+            mock_remove.side_effect = OSError
+            mock_atexit.call_args[0][0]()  # call _cleanup_temp_files
+            mock_remove.assert_has_calls([call('1.tmp'), call('2.tmp')])
+
 
 @patch('urllib3.PoolManager.request')
 class TestApiClient(unittest.TestCase):
@@ -223,6 +238,13 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         with patch.object(Kubernetes, '_wait_caches', Mock(side_effect=Exception)):
             self.assertRaises(KubernetesError, self.k.get_cluster)
 
+    def test_attempt_to_acquire_leader(self):
+        with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', create=True) as mock_patch:
+            mock_patch.side_effect = K8sException
+            self.assertRaises(KubernetesError, self.k.attempt_to_acquire_leader)
+            mock_patch.side_effect = k8s_client.rest.ApiException(409, '')
+            self.assertFalse(self.k.attempt_to_acquire_leader())
+
     def test_take_leader(self):
         self.k.take_leader()
         self.k._leader_observed_record['leader'] = 'test'
@@ -278,7 +300,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
 
     @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
     def test_update_leader(self, mock_patch_namespaced_endpoints):
-        self.assertIsNotNone(self.k.update_leader('123'))
+        self.assertIsNotNone(self.k.update_leader('123', failsafe={'foo': 'bar'}))
         args = mock_patch_namespaced_endpoints.call_args[0]
         self.assertEqual(args[2].subsets[0].addresses[0].target_ref.resource_version, '10')
         self.k._kinds._object_cache['test'].subsets[:] = []
@@ -293,7 +315,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
         mock_patch.side_effect = k8s_client.rest.ApiException(502, '')
         self.assertFalse(self.k.update_leader('123'))
         mock_patch.side_effect = RetryFailedError('')
-        self.assertFalse(self.k.update_leader('123'))
+        self.assertRaises(KubernetesError, self.k.update_leader, '123')
         mock_patch.side_effect = k8s_client.rest.ApiException(409, '')
         with patch('time.time', Mock(side_effect=[0, 100, 200, 0, 0, 0, 0, 100, 200])):
             self.assertFalse(self.k.update_leader('123'))
@@ -303,6 +325,8 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
         mock_read.return_value.metadata.resource_version = '2'
         self.assertIsNotNone(self.k._update_leader_with_retry({}, '1', []))
         mock_patch.side_effect = k8s_client.rest.ApiException(409, '')
+        mock_read.side_effect = RetryFailedError('')
+        self.assertRaises(KubernetesError, self.k.update_leader, '123')
         mock_read.side_effect = Exception
         self.assertFalse(self.k.update_leader('123'))
 

tests/test_postgresql.py

@@ -640,7 +640,7 @@ class TestPostgresql(BaseTestPostgresql):
 
     def test_pick_sync_standby(self):
         cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
-                          SyncState(0, self.me.name, self.leadermem.name), None, None)
+                          SyncState(0, self.me.name, self.leadermem.name), None, None, None)
         mock_cursor = Mock()
         mock_cursor.fetchone.return_value = ('remote_apply',)
 

tests/test_raft.py

@@ -4,7 +4,7 @@ import tempfile
 import time
 
 from mock import Mock, PropertyMock, patch
-from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility, TCPTransport, _TCPTransport
+from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
 from pysyncobj import SyncObjConf, FAIL_REASON
 
 
@@ -79,6 +79,9 @@ class TestKVStoreTTL(unittest.TestCase):
         self.assertFalse(self.so.set('foo', 'bar', prevExist=False, ttl=30))
         self.assertFalse(self.so.retry(self.so._set, 'foo', {'value': 'buz', 'created': 1, 'updated': 1}, prevValue=''))
         self.assertTrue(self.so.retry(self.so._set, 'foo', {'value': 'buz', 'created': 1, 'updated': 1}))
+        with patch.object(KVStoreTTL, 'retry', Mock(side_effect=RaftError(''))):
+            self.assertFalse(self.so.set('foo', 'bar'))
+            self.assertRaises(RaftError, self.so.set, 'foo', 'bar', handle_raft_error=False)
 
     def test_delete(self):
         self.so.autoTickPeriod = 0.2
@@ -87,6 +90,8 @@ class TestKVStoreTTL(unittest.TestCase):
         self.assertFalse(self.so.delete('foo', prevValue='buz'))
         self.assertTrue(self.so.delete('foo', recursive=True))
         self.assertFalse(self.so.retry(self.so._delete, 'foo', prevValue=''))
+        with patch.object(KVStoreTTL, 'retry', Mock(side_effect=RaftError(''))):
+            self.assertFalse(self.so.delete('foo'))
 
     def test_expire(self):
         self.so.set('foo', 'bar', ttl=0.001)
@@ -102,7 +107,7 @@ class TestKVStoreTTL(unittest.TestCase):
             callback(True, return_values.pop(0))
 
         with patch('time.time', Mock(side_effect=[1, 100])):
-            self.assertFalse(self.so.retry(test))
+            self.assertRaises(RaftError, self.so.retry, test)
 
         self.assertTrue(self.so.retry(test))
         self.assertFalse(self.so.retry(test))
@@ -135,7 +140,8 @@ class TestRaft(unittest.TestCase):
         raft.get_cluster()
         self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}'))
         raft.get_cluster()
-        self.assertTrue(raft.update_leader('1'))
+        self.assertTrue(raft.update_leader('1', failsafe={'foo': 'bat'}))
+        self.assertTrue(raft._sync_obj.set(raft.failsafe_path, '{"foo"}'))
         self.assertTrue(raft._sync_obj.set(raft.status_path, '{'))
         raft.get_cluster()
         self.assertTrue(raft.delete_sync_state())

tests/test_slots.py

@@ -31,14 +31,14 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.start()
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
         self.cluster = Cluster(True, config, self.leader, 0,
-                               [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
+                               [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345}, None)
 
     def test_sync_replication_slots(self):
         config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
                                              'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
         cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10})
+                          [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10}, None)
         with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
             self.s.sync_replication_slots(cluster, False)
         self.p.set_role('standby_leader')
@@ -67,7 +67,8 @@ class TestSlotsHandler(BaseTestPostgresql):
     def test_process_permanent_slots(self):
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
-        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None, None)
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem], None, None, None, None, None)
 
         self.s.sync_replication_slots(cluster, False)
         with patch.object(Postgresql, '_query') as mock_query:

tests/test_watchdog.py

@@ -164,6 +164,13 @@ class TestWatchdog(unittest.TestCase):
 
         watchdog.reload_config({'ttl': 60, 'loop_wait': 15, 'watchdog': {'mode': 'required'}})
         watchdog.keepalive()
+        self.assertTrue(watchdog.is_running)
+        self.assertEqual(watchdog.config.timeout, 60 - 5)
+
+        watchdog.reload_config({'ttl': 60, 'loop_wait': 15, 'watchdog': {'mode': 'required', 'safety_margin': -1}})
+        watchdog.keepalive()
+        self.assertTrue(watchdog.is_running)
+        self.assertEqual(watchdog.config.timeout, 60 // 2)
 
 
 class TestNullWatchdog(unittest.TestCase):

tests/test_zookeeper.py

@@ -6,6 +6,7 @@ from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.handlers.threading import SequentialThreadingHandler
 from kazoo.protocol.states import KeeperState, ZnodeStat
+from kazoo.retry import RetryFailedError
 from mock import Mock, PropertyMock, patch
 from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\
         PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
@@ -50,6 +51,8 @@ class MockKazooClient(Mock):
             return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
         elif path.endswith('/status'):
             return (b'{"optime":500,"slots":{"ls":1234567}}', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
+        elif path.endswith('/failsafe'):
+            return (b'{a}', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
         return (b'', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
 
     @staticmethod
@@ -59,7 +62,7 @@ class MockKazooClient(Mock):
         if path.startswith('/no_node'):
             raise NoNodeError
         elif path in ['/service/bla/', '/service/test/']:
-            return ['initialize', 'leader', 'members', 'optime', 'failover', 'sync']
+            return ['initialize', 'leader', 'members', 'optime', 'failover', 'sync', 'failsafe']
         return ['foo', 'bar', 'buzz']
 
     def create(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False):
@@ -173,7 +176,6 @@ class TestZooKeeper(unittest.TestCase):
         self.zk._inner_load_cluster()
 
     def test_get_cluster(self):
-        self.assertRaises(ZooKeeperError, self.zk.get_cluster)
         cluster = self.zk.get_cluster(True)
         self.assertIsInstance(cluster.leader, Leader)
         self.zk.status_watcher(None)
@@ -222,13 +224,25 @@ class TestZooKeeper(unittest.TestCase):
         self.zk.touch_member({'conn_url': 'postgres://repuser:rep-pass@localhost:5434/postgres',
                               'api_url': 'http://127.0.0.1:8009/patroni'})
 
+    @patch.object(MockKazooClient, 'create', Mock(side_effect=[RetryFailedError, Exception]))
+    def test_attempt_to_acquire_leader(self):
+        self.assertRaises(ZooKeeperError, self.zk.attempt_to_acquire_leader)
+        self.assertFalse(self.zk.attempt_to_acquire_leader())
+
     def test_take_leader(self):
         self.zk.take_leader()
         with patch.object(MockKazooClient, 'create', Mock(side_effect=Exception)):
             self.zk.take_leader()
 
     def test_update_leader(self):
-        self.assertTrue(self.zk.update_leader(12345))
+        self.assertFalse(self.zk.update_leader(12345))
+        with patch.object(MockKazooClient, 'delete', Mock(side_effect=RetryFailedError)):
+            self.assertRaises(ZooKeeperError, self.zk.update_leader, 12345)
+        with patch.object(MockKazooClient, 'delete', Mock(side_effect=NoNodeError)):
+            self.assertTrue(self.zk.update_leader(12345, failsafe={'foo': 'bar'}))
+            with patch.object(MockKazooClient, 'create', Mock(side_effect=[RetryFailedError, Exception])):
+                self.assertRaises(ZooKeeperError, self.zk.update_leader, 12345)
+                self.assertFalse(self.zk.update_leader(12345))
 
     @patch.object(Cluster, 'min_version', PropertyMock(return_value=(2, 0)))
     def test_write_leader_optime(self):