Release 2.1.2 (#2136 )

* Implement missing unit-tests * Bump version * Update release notes
Handle missing timelines in history file when deciding to rewind (#2120 )
2026-08-26 07:30:14 +00:00 · 2021-12-03 15:49:57 +01:00 · 2021-12-02 11:35:30 +01:00 · 2021-12-02 11:35:11 +01:00 · 2021-11-30 16:33:38 +01:00 · 2021-11-30 14:20:55 +01:00
44 changed files with 660 additions and 282 deletions
@@ -18,7 +18,10 @@ def install_requirements(what):
    finally:
        sys.path = old_path
    requirements = ['mock>=2.0.0', 'flake8', 'pytest', 'pytest-cov'] if what == 'all' else ['behave']
-    requirements += ['psycopg2-binary', 'coverage']
+    requirements += ['coverage']
+    # try to split tests between psycopg2 and psycopg3
+    requirements += ['psycopg[binary]'] if sys.version_info >= (3, 6, 0) and\
+        (sys.platform != 'darwin' or what == 'etcd3') else ['psycopg2-binary']
    for r in read('requirements.txt').split('\n'):
        r = r.strip()
        if r != '':
@@ -12,7 +12,7 @@ Patroni is a template for you to create your own customized, high-availability s

 We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely.

-Currently supported PostgreSQL versions: 9.3 to 13.
+Currently supported PostgreSQL versions: 9.3 to 14.

 **Note to Kubernetes users**: Patroni can run natively on top of Kubernetes. Take a look at the `Kubernetes <https://github.com/zalando/patroni/blob/master/docs/kubernetes.rst>`__ chapter of the Patroni documentation.

@@ -61,7 +61,7 @@ To install requirements on a Mac, run the following:

    brew install postgresql etcd haproxy libyaml python

-**Psycopg2**
+**Psycopg**

 Starting from `psycopg2-2.8 <http://initd.org/psycopg/articles/2019/04/04/psycopg-28-released/>`__ the binary version of psycopg2 will no longer be installed by default. Installing it from the source code requires C compiler and postgres+python dev packages.
 Since in the python world it is not possible to specify dependency as ``psycopg2 OR psycopg2-binary`` you will have to decide how to install it.
@@ -88,6 +88,12 @@ There are a few options available:

    pip install psycopg2>=2.5.4

+4. Use psycopg 3.0 instead of psycopg2
+
+::
+
+    pip install psycopg[binary]
+
 **General installation for pip**

 Patroni can be installed with pip:
@@ -84,6 +84,7 @@ ZooKeeper
 -  **PATRONI\_ZOOKEEPER\_KEY**: (optional) File with the client key.
 -  **PATRONI\_ZOOKEEPER\_KEY\_PASSWORD**: (optional) The client key password.
 -  **PATRONI\_ZOOKEEPER\_VERIFY**: (optional) Whether to verify certificate or not. Defaults to ``true``.
+-  **PATRONI\_ZOOKEEPER\_SET\_ACLS**: (optional) If set, configure Kazoo to apply a default ACL to each ZNode that it creates. ACLs will assume 'x509' schema and should be specified as a dictionary with the principal as the key and one or more permissions as a list in the value.  Permissions may be one of ``CREATE``, ``READ``, ``WRITE``, ``DELETE`` or ``ADMIN``.  For example, ``set_acls: {CN=principal1: [CREATE, READ], CN=principal2: [ALL]}``.

 .. note::
    It is required to install ``kazoo>=2.6.0`` to support SSL.
@@ -133,6 +134,7 @@ PostgreSQL
 -  **PATRONI\_REPLICATION\_SSLCERT**: (optional) maps to the `sslcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCERT>`__ connection parameter, which specifies the location of the client certificate.
 -  **PATRONI\_REPLICATION\_SSLROOTCERT**: (optional) maps to the `sslrootcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLROOTCERT>`__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate.
 -  **PATRONI\_REPLICATION\_SSLCRL**: (optional) maps to the `sslcrl <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRL>`__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
+-  **PATRONI\_REPLICATION\_SSLCRLDIR**: (optional) maps to the `sslcrldir <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRLDIR>`__ connection parameter, which specifies the location of a directory with files containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
 -  **PATRONI\_REPLICATION\_GSSENCMODE**: (optional) maps to the `gssencmode <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-GSSENCMODE>`__ connection parameter, which determines whether or with what priority a secure GSS TCP/IP connection will be negotiated with the server
 -  **PATRONI\_REPLICATION\_CHANNEL\_BINDING**: (optional) maps to the `channel_binding <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-CHANNEL-BINDING>`__ connection parameter, which controls the client's use of channel binding.
 -  **PATRONI\_SUPERUSER\_USERNAME**: name for the superuser, set during initialization (initdb) and later used by Patroni to connect to the postgres. Also this user is used by pg_rewind.
@@ -143,6 +145,7 @@ PostgreSQL
 -  **PATRONI\_SUPERUSER\_SSLCERT**: (optional) maps to the `sslcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCERT>`__ connection parameter, which specifies the location of the client certificate.
 -  **PATRONI\_SUPERUSER\_SSLROOTCERT**: (optional) maps to the `sslrootcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLROOTCERT>`__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate.
 -  **PATRONI\_SUPERUSER\_SSLCRL**: (optional) maps to the `sslcrl <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRL>`__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
+-  **PATRONI\_SUPERUSER\_SSLCRLDIR**: (optional) maps to the `sslcrldir <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRLDIR>`__ connection parameter, which specifies the location of a directory with files containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
 -  **PATRONI\_SUPERUSER\_GSSENCMODE**: (optional) maps to the `gssencmode <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-GSSENCMODE>`__ connection parameter, which determines whether or with what priority a secure GSS TCP/IP connection will be negotiated with the server
 -  **PATRONI\_SUPERUSER\_CHANNEL\_BINDING**: (optional) maps to the `channel_binding <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-CHANNEL-BINDING>`__ connection parameter, which controls the client's use of channel binding.
 -  **PATRONI\_REWIND\_USERNAME**: name for the user for ``pg_rewind``; the user will be created during initialization of postgres 11+ and all necessary `permissions <https://www.postgresql.org/docs/11/app-pgrewind.html#id-1.9.5.8.8>`__ will be granted.
@@ -153,6 +156,7 @@ PostgreSQL
 -  **PATRONI\_REWIND\_SSLCERT**: (optional) maps to the `sslcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCERT>`__ connection parameter, which specifies the location of the client certificate.
 -  **PATRONI\_REWIND\_SSLROOTCERT**: (optional) maps to the `sslrootcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLROOTCERT>`__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate.
 -  **PATRONI\_REWIND\_SSLCRL**: (optional) maps to the `sslcrl <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRL>`__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
+-  **PATRONI\_REWIND\_SSLCRLDIR**: (optional) maps to the `sslcrldir <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRLDIR>`__ connection parameter, which specifies the location of a directory with files containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
 -  **PATRONI\_REWIND\_GSSENCMODE**: (optional) maps to the `gssencmode <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-GSSENCMODE>`__ connection parameter, which determines whether or with what priority a secure GSS TCP/IP connection will be negotiated with the server
 -  **PATRONI\_REWIND\_CHANNEL\_BINDING**: (optional) maps to the `channel_binding <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-CHANNEL-BINDING>`__ connection parameter, which controls the client's use of channel binding.

@@ -35,7 +35,7 @@ To install requirements on a Mac, run the following:

 .. _psycopg2_install_options:

-**Psycopg2**
+**Psycopg**

 Starting from `psycopg2-2.8 <http://initd.org/psycopg/articles/2019/04/04/psycopg-28-released/>`__ the binary version of psycopg2 will no longer be installed by default. Installing it from the source code requires C compiler and postgres+python dev packages.
 Since in the python world it is not possible to specify dependency as ``psycopg2 OR psycopg2-binary`` you will have to decide how to install it.
@@ -62,6 +62,12 @@ There are a few options available:

    pip install psycopg2>=2.5.4

+4. Use psycopg 3.0 instead of psycopg2
+
+::
+
+    pip install psycopg[binary]>=3.0.0
+
 **General installation for pip**

 Patroni can be installed with pip:
@@ -182,6 +182,7 @@ ZooKeeper
 -  **key**: (optional) File with the client key.
 -  **key_password**: (optional) The client key password.
 -  **verify**: (optional) Whether to verify certificate or not. Defaults to ``true``.
+-  **set_acls**: (optional) If set, configure Kazoo to apply a default ACL to each ZNode that it creates. ACLs will assume 'x509' schema and should be specified as a dictionary with the principal as the key and one or more permissions as a list in the value.  Permissions may be one of ``CREATE``, ``READ``, ``WRITE``, ``DELETE`` or ``ADMIN``.  For example, ``set_acls: {CN=principal1: [CREATE, READ], CN=principal2: [ALL]}``. 

 .. note::
    It is required to install ``kazoo>=2.6.0`` to support SSL.
@@ -256,6 +257,7 @@ PostgreSQL
            -  **sslcert**: (optional) maps to the `sslcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCERT>`__ connection parameter, which specifies the location of the client certificate.
            -  **sslrootcert**: (optional) maps to the `sslrootcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLROOTCERT>`__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate.
            -  **sslcrl**: (optional) maps to the `sslcrl <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRL>`__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
+            -  **sslcrldir**: (optional) maps to the `sslcrldir <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRLDIR>`__ connection parameter, which specifies the location of a directory with files containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
            -  **gssencmode**: (optional) maps to the `gssencmode <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-GSSENCMODE>`__ connection parameter, which determines whether or with what priority a secure GSS TCP/IP connection will be negotiated with the server
            -  **channel_binding**: (optional) maps to the `channel_binding <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-CHANNEL-BINDING>`__ connection parameter, which controls the client's use of channel binding.
        -  **replication**:
@@ -267,6 +269,7 @@ PostgreSQL
            -  **sslcert**: (optional) maps to the `sslcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCERT>`__ connection parameter, which specifies the location of the client certificate.
            -  **sslrootcert**: (optional) maps to the `sslrootcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLROOTCERT>`__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate.
            -  **sslcrl**: (optional) maps to the `sslcrl <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRL>`__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
+            -  **sslcrldir**: (optional) maps to the `sslcrldir <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRLDIR>`__ connection parameter, which specifies the location of a directory with files containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
            -  **gssencmode**: (optional) maps to the `gssencmode <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-GSSENCMODE>`__ connection parameter, which determines whether or with what priority a secure GSS TCP/IP connection will be negotiated with the server
            -  **channel_binding**: (optional) maps to the `channel_binding <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-CHANNEL-BINDING>`__ connection parameter, which controls the client's use of channel binding.
        -  **rewind**:
@@ -278,6 +281,7 @@ PostgreSQL
            -  **sslcert**: (optional) maps to the `sslcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCERT>`__ connection parameter, which specifies the location of the client certificate.
            -  **sslrootcert**: (optional) maps to the `sslrootcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLROOTCERT>`__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate.
            -  **sslcrl**: (optional) maps to the `sslcrl <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRL>`__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
+            -  **sslcrldir**: (optional) maps to the `sslcrldir <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRLDIR>`__ connection parameter, which specifies the location of a directory with files containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
            -  **gssencmode**: (optional) maps to the `gssencmode <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-GSSENCMODE>`__ connection parameter, which determines whether or with what priority a secure GSS TCP/IP connection will be negotiated with the server
            -  **channel_binding**: (optional) maps to the `channel_binding <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-CHANNEL-BINDING>`__ connection parameter, which controls the client's use of channel binding.
    -  **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.)
@@ -194,4 +194,7 @@ intersphinx_mapping = {'https://docs.python.org/': None}
 # A possibility to have an own stylesheet, to add new rules or override existing ones
 # For the latter case, the CSS specificity of the rules should be higher than the default ones
 def setup(app):
-   app.add_stylesheet("custom.css")
+    if hasattr(app, 'add_css_file'):
+        app.add_css_file('custom.css')
+    else:
+        app.add_stylesheet('custom.css')
@@ -10,7 +10,7 @@ Patroni is a template for you to create your own customized, high-availability s

 We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely. There are many ways to run high availability with PostgreSQL; for a list, see the `PostgreSQL Documentation <https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling>`__.

-Currently supported PostgreSQL versions: 9.3 to 13.
+Currently supported PostgreSQL versions: 9.3 to 14.

 **Note to Kubernetes users**: Patroni can run natively on top of Kubernetes. Take a look at the :ref:`Kubernetes <kubernetes>` chapter of the Patroni documentation.

@@ -3,6 +3,110 @@
 Release notes
 =============

+Version 2.1.2
+-------------
+
+**New features**
+
+- Compatibility with ``psycopg>=3.0`` (Alexander Kukushkin)
+
+  By default ``psycopg2`` is preferred. `psycopg>=3.0` will be used only if ``psycopg2`` is not available or its version is too old.
+
+- Add ``dcs_last_seen`` field to the REST API (Michael Banck)
+
+  This field notes the last time (as unix epoch) a cluster member has successfully communicated with the DCS. This is useful to identify and/or analyze network partitions.
+
+- Release the leader lock when ``pg_controldata`` reports "shut down" (Alexander)
+
+  To solve the problem of slow switchover/shutdown in case ``archive_command`` is slow/failing, Patroni will remove the leader key immediately after ``pg_controldata`` started reporting PGDATA as ``shut down`` cleanly and it verified that there is at least one replica that received all changes. If there are no replicas that fulfill this condition the leader key is not removed and the old behavior is retained, i.e. Patroni will keep updating the lock.
+
+- Add ``sslcrldir`` connection parameter support (Kostiantyn Nemchenko)
+
+  The new connection parameter was introduced in the PostgreSQL v14.
+
+- Allow setting ACLs for ZNodes in Zookeeper (Alwyn Davis)
+
+  Introduce a new configuration option ``zookeeper.set_acls`` so that Kazoo will apply a default ACL for each ZNode that it creates.
+
+
+**Stability improvements**
+
+- Delay the next attempt of recovery till next HA loop (Alexander)
+
+  If Postgres crashed due to out of disk space (for example) and fails to start because of that Patroni is too eagerly trying to recover it flooding logs.
+
+- Add log before demoting, which can take some time (Michael)
+
+  It can take some time for the demote to finish and it might not be obvious from looking at the logs what exactly is going on.
+
+- Improve "I am" status messages (Michael)
+
+  ``no action. I am a secondary ({0})`` vs ``no action. I am ({0}), a secondary``
+
+- Cast to int ``wal_keep_segments`` when converting to ``wal_keep_size`` (Jorge Solórzano)
+
+  It is possible to specify ``wal_keep_segments`` as a string in the global :ref:`dynamic configuration <dynamic_configuration>` and due to Python being a dynamically typed language the string was simply multiplied. Example: ``wal_keep_segments: "100"`` was converted to ``100100100100100100100100100100100100100100100100MB``.
+
+- Allow switchover only to sync nodes when synchronous replication is enabled (Alexander)
+
+  In addition to that do the leader race only against known synchronous nodes.
+
+- Use cached role as a fallback when Postgres is slow (Alexander)
+
+  In some extreme cases Postgres could be so slow that the normal monitoring query does not finish in a few seconds. The ``statement_timeout`` exception not being properly handled could lead to the situation where Postgres was not demoted on time when the leader key expired or the update failed. In case of such exception Patroni will use the cached ``role`` to determine whether Postgres is running as a primary.
+
+- Avoid unnecessary updates of the member ZNode (Alexander)
+
+  If no values have changed in the members data, the update should not happen.
+
+- Optimize checkpoint after promote (Alexander)
+
+  Avoid doing ``CHECKPOINT`` if the latest timeline is already stored in ``pg_control``. It helps to avoid unnecessary ``CHECKPOINT`` right after initializing the new cluster with ``initdb``.
+
+- Prefer members without ``nofailover`` when picking sync nodes (Alexander)
+
+  Previously sync nodes were selected only based on the replication lag, hence the node with ``nofailover`` tag had the same chances to become synchronous as any other node. That behavior was confusing and dangerous at the same time because in case of a failed primary the failover could not happen automatically.
+
+- Remove duplicate hosts from the etcd machine cache (Michael)
+
+  Advertised client URLs in the etcd cluster could be misconfigured. Removing duplicates in Patroni in this case is a low-hanging fruit.
+
+
+**Bugfixes**
+
+- Skip temporary replication slots while doing slot management (Alexander)
+
+  Starting from v10 ``pg_basebackup`` creates a temporary replication slot for WAL streaming and Patroni was trying to drop it because the slot name looks unknown. In order to fix it, we skip all temporary slots when querying ``pg_stat_replication_slots`` view.
+
+- Ensure ``pg_replication_slot_advance()`` doesn't timeout (Alexander)
+
+  Patroni was using the default ``statement_timeout`` in this case and once the call failed there are very high chances that it will never recover, resulting in increased size of ``pg_wal`` and ``pg_catalog`` bloat.
+
+- The ``/status`` wasn't updated on demote (Alexander)
+
+  After demoting PostgreSQL the old leader updates the last LSN in DCS. Starting from ``2.1.0`` the new ``/status`` key was introduced, but the optime was still written to the ``/optime/leader``.
+
+- Handle DCS exceptions when demoting (Alexander)
+
+  While demoting the master due to failure to update the leader lock it could happen that DCS goes completely down and the ``get_cluster()`` call raises an exception. Not being handled properly it results in Postgres remaining stopped until DCS recovers.
+
+- The ``use_unix_socket_repl`` didn't work is some cases (Alexander)
+
+  Specifically, if ``postgresql.unix_socket_directories`` is not set. In this case Patroni is supposed to use the default value from ``libpq``.
+
+- Fix a few issues with Patroni REST API (Alexander)
+
+  The ``clusters_unlocked`` sometimes could be not defined, what resulted in exceptions in the ``GET /metrics`` endpoint. In addition to that the error handling method was assuming that the ``connect_address`` tuple always has two elements, while in fact there could be more in case of IPv6.
+
+- Wait for newly promoted node to finish recovery before deciding to rewind (Alexander)
+
+  It could take some time before the actual promote happens and the new timeline is created. Without waiting replicas could come to the conclusion that rewind isn't required.
+
+- Handle missing timelines in a history file when deciding to rewind (Alexander)
+
+  If the current replica timeline is missing in the history file on the primary the replica was falsely assuming that rewind isn't required.
+
+
 Version 2.1.1
 -------------

@@ -1,7 +1,6 @@
 import abc
 import datetime
 import os
-import psycopg2
 import json
 import shutil
 import signal
@@ -13,6 +12,8 @@ import threading
 import time
 import yaml

+import patroni.psycopg as psycopg
+
 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer


@@ -205,16 +206,16 @@ class PatroniController(AbstractController):

        user = 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, 'database': 'postgres'})
+        self._connkwargs.update({'host': host, 'port': self.__PORT, 'dbname': 'postgres'})

        self._replication = config['postgresql'].get('authentication', config['postgresql']).get('replication', {})
-        self._replication.update({'host': host, 'port': self.__PORT, 'database': 'postgres'})
+        self._replication.update({'host': host, 'port': self.__PORT, 'dbname': 'postgres'})

        return patroni_config_path

    def _connection(self):
        if not self._conn or self._conn.closed != 0:
-            self._conn = psycopg2.connect(**self._connkwargs)
+            self._conn = psycopg.connect(**self._connkwargs)
            self._conn.autocommit = True
        return self._conn

@@ -228,7 +229,7 @@ class PatroniController(AbstractController):
            cursor = self._cursor()
            cursor.execute(query)
            return cursor
-        except psycopg2.Error:
+        except psycopg.Error:
            if not fail_ok:
                raise

@@ -268,7 +269,7 @@ class PatroniController(AbstractController):

    @property
    def backup_source(self):
-        return 'postgres://{username}:{password}@{host}:{port}/{database}'.format(**self._replication)
+        return 'postgres://{username}:{password}@{host}:{port}/{dbname}'.format(**self._replication)

    def backup(self, dest=os.path.join('data', 'basebackup')):
        subprocess.call(PatroniPoolController.BACKUP_SCRIPT + ['--walmethod=none',
@@ -71,6 +71,7 @@ Scenario: check API requests for the primary-replica pair in the pause mode
 	When I run patronictl.py restart batman postgres1 --force
 	Then I receive a response returncode 0
 	Then replication works from postgres0 to postgres1 after 20 seconds
+	And I sleep for 2 seconds
 	When I issue a GET request to http://127.0.0.1:8009/replica
 	Then I receive a response code 200
 	And I receive a response state running
@@ -1,4 +1,4 @@
-import psycopg2 as pg
+import patroni.psycopg as pg

 from behave import step, then
 from time import sleep, time
@@ -1,7 +1,7 @@
 import time
-import psycopg2

 from behave import step, then
+import patroni.psycopg as pg


@step('I create a logical replication slot {slot_name} on {pg_name:w} with the {plugin:w} plugin')
@@ -10,7 +10,7 @@ def create_logical_replication_slot(context, slot_name, pg_name, plugin):
        output = context.pctl.query(pg_name, ("SELECT pg_create_logical_replication_slot('{0}', '{1}'),"
                                              " current_database()").format(slot_name, plugin))
        print(output.fetchone())
-    except psycopg2.Error as e:
+    except pg.Error as e:
        print(e)
        assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)

@@ -24,7 +24,7 @@ def has_logical_replication_slot(context, pg_name, slot_name, plugin):
        assert row[0] == "logical", "Found replication slot named {0} but wasn't a logical slot".format(slot_name)
        assert row[1] == plugin, ("Found replication slot named {0} but was using plugin "
                                  "{1} rather than {2}").format(slot_name, row[1], plugin)
-    except psycopg2.Error:
+    except pg.Error:
        assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)


@@ -34,7 +34,7 @@ def does_not_have_logical_replication_slot(context, pg_name, slot_name):
        row = context.pctl.query(pg_name, ("SELECT 1 FROM pg_replication_slots"
                                           " WHERE slot_name = '{0}'").format(slot_name)).fetchone()
        assert not row, "Found unexpected replication slot named {0}".format(slot_name)
-    except psycopg2.Error:
+    except pg.Error:
        assert False, "Error looking for slot {0} on {1}".format(slot_name, pg_name)


@@ -11,6 +11,7 @@ logger = logging.getLogger(__name__)

 PATRONI_ENV_PREFIX = 'PATRONI_'
 KUBERNETES_ENV_PREFIX = 'KUBERNETES_'
+MIN_PSYCOPG2 = (2, 5, 4)


 class Patroni(AbstractPatroniDaemon):
@@ -144,30 +145,40 @@ def fatal(string, *args):
    sys.exit(1)


-def check_psycopg2():
-    min_psycopg2 = (2, 5, 4)
-    min_psycopg2_str = '.'.join(map(str, min_psycopg2))
-
-    def parse_version(version):
+def parse_version(version):
+    def _parse_version(version):
        for e in version.split('.'):
            try:
                yield int(e)
            except ValueError:
                break
+    return tuple(_parse_version(version.split(' ')[0]))
+
+
+# We pass MIN_PSYCOPG2 and parse_version as arguments to simplify usage of check_psycopg from the setup.py
+def check_psycopg(_min_psycopg2=MIN_PSYCOPG2, _parse_version=parse_version):
+    min_psycopg2_str = '.'.join(map(str, _min_psycopg2))

    try:
-        import psycopg2
-        version_str = psycopg2.__version__.split(' ')[0]
-        version = tuple(parse_version(version_str))
-        if version < min_psycopg2:
-            fatal('Patroni requires psycopg2>={0}, but only {1} is available', min_psycopg2_str, version_str)
+        from psycopg2 import __version__
+        if _parse_version(__version__) >= _min_psycopg2:
+            return
+        version_str = __version__.split(' ')[0]
    except ImportError:
-        fatal('Patroni requires psycopg2>={0} or psycopg2-binary', min_psycopg2_str)
+        version_str = None
+
+    try:
+        from psycopg import __version__
+    except ImportError:
+        error = 'Patroni requires psycopg2>={0}, psycopg2-binary, or psycopg>=3.0'.format(min_psycopg2_str)
+        if version_str:
+            error += ', but only psycopg2=={0} is available'.format(version_str)
+        fatal(error)


 def main():
    if os.getpid() != 1:
-        check_psycopg2()
+        check_psycopg()
        return patroni_main()

    # Patroni started with PID=1, it looks like we are in the container
@@ -2,7 +2,6 @@ import base64
 import hmac
 import json
 import logging
-import psycopg2
 import time
 import traceback
 import dateutil.parser
@@ -18,6 +17,7 @@ from six.moves.socketserver import ThreadingMixIn
 from six.moves.urllib_parse import urlparse, parse_qs
 from threading import Thread

+from . import psycopg
 from .exceptions import PostgresConnectionException, PostgresException
 from .postgresql.misc import postgres_version_to_int
 from .utils import deep_compare, enable_keepalive, parse_bool, patch_config, Retry, \
@@ -282,12 +282,17 @@ class RestApiHandler(BaseHTTPRequestHandler):

        metrics.append("# HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked.")
        metrics.append("# TYPE patroni_cluster_unlocked gauge")
-        metrics.append("patroni_cluster_unlocked{0} {1}".format(scope_label, int(postgres['cluster_unlocked'])))
+        metrics.append("patroni_cluster_unlocked{0} {1}".format(scope_label, int(postgres.get('cluster_unlocked', 0))))

        metrics.append("# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.")
        metrics.append("# TYPE patroni_postgres_timeline counter")
        metrics.append("patroni_postgres_timeline{0} {1}".format(scope_label, postgres.get('timeline', 0)))

+        metrics.append("# HELP patroni_dcs_last_seen Epoch timestamp when DCS was last contacted successfully"
+                       " by Patroni.")
+        metrics.append("# TYPE patroni_dcs_last_seen gauge")
+        metrics.append("patroni_dcs_last_seen{0} {1}".format(scope_label, postgres.get('dcs_last_seen', 0)))
+
        self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain')

    def _read_json_content(self, body_is_optional=False):
@@ -599,7 +604,6 @@ class RestApiHandler(BaseHTTPRequestHandler):
                'postmaster_start_time': row[0],
                'role': 'replica' if row[1] == 0 else 'master',
                'server_version': postgresql.server_version,
-                'cluster_unlocked': bool(not cluster or cluster.is_unlocked()),
                'xlog': ({
                    'received_location': row[4] or row[3],
                    'replayed_location': row[3],
@@ -621,13 +625,17 @@ class RestApiHandler(BaseHTTPRequestHandler):
            if row[7]:
                result['replication'] = row[7]

-            return result
-        except (psycopg2.Error, RetryFailedError, PostgresConnectionException):
+        except (psycopg.Error, RetryFailedError, PostgresConnectionException):
            state = postgresql.state
            if state == 'running':
                logger.exception('get_postgresql_status')
                state = 'unknown'
-            return {'state': state, 'role': postgresql.role}
+            result = {'state': state, 'role': postgresql.role}
+
+        if not cluster or cluster.is_unlocked():
+            result['cluster_unlocked'] = True
+        result['dcs_last_seen'] = self.server.patroni.dcs.last_seen
+        return result

    def handle_one_request(self):
        self.__start_time = time.time()
@@ -657,7 +665,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
            with self.patroni.postgresql.connection().cursor() as cursor:
                cursor.execute(sql, params)
                return [r for r in cursor]
-        except psycopg2.Error as e:
+        except psycopg.Error as e:
            if cursor and cursor.connection.closed == 0:
                raise e
            raise PostgresConnectionException('connection problems')
@@ -867,6 +875,6 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):

    @staticmethod
    def handle_error(request, client_address):
-        address, port = client_address
-        logger.warning('Exception happened during processing of request from {}:{}'.format(address, port))
+        logger.warning('Exception happened during processing of request from %s:%s',
+                       client_address[0], client_address[1])
        logger.warning(traceback.format_exc())
@@ -24,6 +24,7 @@ _AUTH_ALLOWED_PARAMETERS = (
    'sslpassword',
    'sslrootcert',
    'sslcrl',
+    'sslcrldir',
    'gssencmode',
    'channel_binding'
 )
@@ -351,13 +352,13 @@ class Config(object):
                              'CACERT', 'CERT', 'KEY', 'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'CONSISTENCY',
                              'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'NAMESPACE', 'CONTEXT',
                              'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS',
-                              'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL') and name:
+                              'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL', 'SET_ACLS') and name:
                    value = os.environ.pop(param)
                    if suffix == 'PORT':
                        value = value and parse_int(value)
                    elif suffix in ('HOSTS', 'PORTS', 'CHECKS'):
                        value = value and _parse_list(value)
-                    elif suffix == 'LABELS':
+                    elif suffix in ('LABELS', 'SET_ACLS'):
                        value = _parse_dict(value)
                    elif suffix in ('USE_PROXIES', 'REGISTER_SERVICE', 'USE_ENDPOINTS', 'BYPASS_API_SERVICE', 'VERIFY'):
                        value = parse_bool(value)
@@ -264,13 +264,13 @@ def get_cursor(cluster, connect_parameters, role='master', member=None):

    params = member.conn_kwargs(connect_parameters)
    params.update({'fallback_application_name': 'Patroni ctl', 'connect_timeout': '5'})
-    if 'database' in connect_parameters:
-        params['database'] = connect_parameters['database']
+    if 'dbname' in connect_parameters:
+        params['dbname'] = connect_parameters['dbname']
    else:
-        params.pop('database')
+        params.pop('dbname')

-    import psycopg2
-    conn = psycopg2.connect(**params)
+    from . import psycopg
+    conn = psycopg.connect(**params)
    conn.autocommit = True
    cursor = conn.cursor()
    if role == 'any':
@@ -401,7 +401,7 @@ def query(
    if password:
        connect_parameters['password'] = click.prompt('Password', hide_input=True, type=str)
    if dbname:
-        connect_parameters['database'] = dbname
+        connect_parameters['dbname'] = dbname

    if p_file is not None:
        command = p_file.read()
@@ -418,7 +418,7 @@ def query(


 def query_member(cluster, cursor, member, role, command, connect_parameters):
-    import psycopg2
+    from . import psycopg
    try:
        if cursor is None:
            cursor = get_cursor(cluster, connect_parameters, role=role, member=member)
@@ -433,11 +433,11 @@ def query_member(cluster, cursor, member, role, command, connect_parameters):

        cursor.execute(command)
        return cursor.fetchall(), [d.name for d in cursor.description]
-    except (psycopg2.OperationalError, psycopg2.DatabaseError) as oe:
-        logging.debug(oe)
+    except psycopg.DatabaseError as de:
+        logging.debug(de)
        if cursor is not None and not cursor.connection.closed:
            cursor.connection.close()
-        message = oe.pgcode or oe.pgerror or str(oe)
+        message = de.diag.sqlstate or str(de)
        message = message.replace('\n', ' ')
        return [[timestamp(0), 'ERROR, SQLSTATE: {0}'.format(message)]], None

@@ -1302,8 +1302,8 @@ def history(obj, cluster_name, fmt):
    table_header_row = ['TL', 'LSN', 'Reason', 'Timestamp', 'New Leader']
    for line in history:
        if len(line) < len(table_header_row):
-            add_coloumn_num = len(table_header_row) - len(line)
-            for _ in range(add_coloumn_num):
+            add_column_num = len(table_header_row) - len(line)
+            for _ in range(add_column_num):
                line.append('')
    print_output(table_header_row, history, {'TL': 'r', 'LSN': 'r'}, fmt)

@@ -160,7 +160,7 @@ class Member(namedtuple('Member', 'index,name,session,data')):
        defaults = {
            "host": None,
            "port": None,
-            "database": None
+            "dbname": None
        }
        ret = self.data.get('conn_kwargs')
        if ret:
@@ -174,7 +174,7 @@ class Member(namedtuple('Member', 'index,name,session,data')):
            ret = {
                'host': r.hostname,
                'port': r.port or 5432,
-                'database': r.path[1:]
+                'dbname': r.path[1:]
            }
            self.data['conn_kwargs'] = ret.copy()

@@ -499,7 +499,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,f

    @property
    def use_slots(self):
-        return self.config and self.config.data.get('postgresql', {}).get('use_slots', True)
+        return self.config and (self.config.data.get('postgresql') or {}).get('use_slots', True)

    def get_replication_slots(self, my_name, role, nofailover, major_version, show_error=False):
        # if the replicatefrom tag is set on the member - we should not create the replication slot for it on
@@ -652,6 +652,7 @@ class AbstractDCS(object):
        self._cluster_valid_till = 0
        self._cluster_thread_lock = Lock()
        self._last_lsn = ''
+        self._last_seen = 0
        self._last_status = {}
        self.event = Event()

@@ -722,6 +723,10 @@ class AbstractDCS(object):
    def loop_wait(self):
        return self._loop_wait

+    @property
+    def last_seen(self):
+        return self._last_seen
+
    @abc.abstractmethod
    def _load_cluster(self):
        """Internally this method should build  `Cluster` object which
@@ -744,6 +749,8 @@ class AbstractDCS(object):
            self.reset_cluster()
            raise

+        self._last_seen = int(time.time())
+
        with self._cluster_thread_lock:
            self._cluster = cluster
            self._cluster_valid_till = time.time() + self.ttl
@@ -767,8 +774,7 @@ class AbstractDCS(object):
        :returns: `!True` on success."""

    def write_leader_optime(self, last_lsn):
-        if self._last_lsn != last_lsn and self._write_leader_optime(last_lsn):
-            self._last_lsn = last_lsn
+        self.write_status({self._OPTIME: last_lsn})

    @abc.abstractmethod
    def _write_status(self, value):
@@ -782,7 +788,8 @@ class AbstractDCS(object):
            self._last_status = value
        cluster = self.cluster
        min_version = cluster and cluster.min_version
-        if min_version and min_version < (2, 1, 0):
+        if min_version and min_version < (2, 1, 0) and self._last_lsn != value[self._OPTIME]:
+            self._last_lsn = value[self._OPTIME]
            self._write_leader_optime(str(value[self._OPTIME]))

    @abc.abstractmethod
@@ -883,7 +890,7 @@ class AbstractDCS(object):
        :param last_lsn: latest checkpoint location in bytes"""

        if last_lsn:
-            self.write_leader_optime(last_lsn)
+            self.write_status({self._OPTIME: last_lsn})
        return self._delete_leader()

    @abc.abstractmethod
@@ -184,7 +184,7 @@ class AbstractEtcdClientWithFailover(etcd.Client):

        for base_uri in machines_cache:
            try:
-                machines = list(self._get_members(base_uri, **kwargs))
+                machines = list(set(self._get_members(base_uri, **kwargs)))
                logger.debug("Retrieved list of machines: %s", machines)
                if machines:
                    random.shuffle(machines)
@@ -345,12 +345,12 @@ class K8sClient(object):
                    try:
                        self._load_api_servers_cache()
                        api_servers_cache = self.api_servers_cache
-                        api_servers = len(api_servers)
+                        api_servers = len(api_servers_cache)
                    except Exception as e:
                        logger.debug('Failed to update list of K8s master nodes: %r', e)

                    sleeptime = retry.sleeptime
-                    remaining_time = retry.stoptime - sleeptime - time.time()
+                    remaining_time = (retry.stoptime or time.time()) - sleeptime - time.time()
                    nodes, timeout, retries = self._calculate_timeouts(api_servers, remaining_time)
                    if nodes == 0:
                        self._update_api_servers_cache = True
@@ -927,7 +927,7 @@ class Kubernetes(AbstractDCS):

        # 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)
+            kind = _retry(self._api.read_namespaced_kind, self.leader_path, self._namespace)
        except Exception as e:
            logger.error('Failed to get the leader object "%s": %r', self.leader_path, e)
            return False
@@ -1011,7 +1011,7 @@ class Kubernetes(AbstractDCS):
    def touch_member(self, data, permanent=False):
        cluster = self.cluster
        if cluster and cluster.leader and cluster.leader.name == self._name:
-            role = 'promoted' if data['role'] in ('replica', 'promoted') else 'master'
+            role = 'master'
        elif data['state'] == 'running' and data['role'] != 'master':
            role = data['role']
        else:
@@ -7,6 +7,7 @@ from kazoo.client import KazooClient, KazooState, KazooRetry
 from kazoo.exceptions import NoNodeError, NodeExistsError, SessionExpiredError
 from kazoo.handlers.threading import SequentialThreadingHandler
 from kazoo.protocol.states import KeeperState
+from kazoo.security import make_acl

 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
 from ..exceptions import DCSError
@@ -83,6 +84,19 @@ class ZooKeeper(AbstractDCS):
                   'cert': 'certfile', 'key': 'keyfile', 'key_password': 'keyfile_password'}
        kwargs = {v: config[k] for k, v in mapping.items() if k in config}

+        if 'set_acls' in config:
+            kwargs['default_acl'] = []
+            for principal, permissions in config['set_acls'].items():
+                normalizedPermissions = [p.upper() for p in permissions]
+                kwargs['default_acl'].append(make_acl(scheme='x509',
+                                                      credential=principal,
+                                                      read='READ' in normalizedPermissions,
+                                                      write='WRITE' in normalizedPermissions,
+                                                      create='CREATE' in normalizedPermissions,
+                                                      delete='DELETE' in normalizedPermissions,
+                                                      admin='ADMIN' in normalizedPermissions,
+                                                      all='ALL' in normalizedPermissions))
+
        self._client = PatroniKazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']),
                                          timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
                                          sleep_func=time.sleep), command_retry=KazooRetry(max_delay=1, max_tries=-1,
@@ -91,6 +105,7 @@ class ZooKeeper(AbstractDCS):

        self._fetch_cluster = True
        self._fetch_status = True
+        self.__last_member_data = None

        self._orig_kazoo_connect = self._client._connection._connect
        self._client._connection._connect = self._kazoo_connect
@@ -268,17 +283,20 @@ class ZooKeeper(AbstractDCS):
                logger.exception('get_cluster')
                self.cluster_watcher(None)
                raise ZooKeeperError('ZooKeeper in not responding properly')
-        # The /status ZNode was updated or doesn't exist and we are not leader
-        elif (self._fetch_status and not self._fetch_cluster or not cluster.last_lsn
-              or cluster.has_permanent_logical_slots(self._name, False) and not cluster.slots) and\
-                not (cluster.leader and cluster.leader.name == self._name):
-            try:
-                last_lsn, slots = self.get_status(cluster.leader)
+        # The /status ZNode was updated or doesn't exist
+        elif self._fetch_status and not self._fetch_cluster or not cluster.last_lsn \
+                or cluster.has_permanent_logical_slots(self._name, False) and not cluster.slots:
+            # If current node is the leader just clear the event without fetching anything (we are updating the /status)
+            if cluster.leader and cluster.leader.name == self._name:
                self.event.clear()
-                cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, last_lsn,
-                                  cluster.members, cluster.failover, cluster.sync, cluster.history, slots)
-            except Exception:
-                pass
+            else:
+                try:
+                    last_lsn, slots = self.get_status(cluster.leader)
+                    self.event.clear()
+                    cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, last_lsn,
+                                      cluster.members, cluster.failover, cluster.sync, cluster.history, slots)
+                except Exception:
+                    pass
        return cluster

    def _bypass_caches(self):
@@ -334,11 +352,11 @@ class ZooKeeper(AbstractDCS):
    def touch_member(self, data, permanent=False):
        cluster = self.cluster
        member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
-        encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8')
+        member_data = self.__last_member_data or member and member.data
        if member and (self._client.client_id is not None and member.session != self._client.client_id[0] or
-                       not (deep_compare(member.data.get('tags', {}), data.get('tags', {})) and
-                            member.data.get('version') == data.get('version') and
-                            member.data.get('checkpoint_after_promote') == data.get('checkpoint_after_promote'))):
+                       not (deep_compare(member_data.get('tags', {}), data.get('tags', {})) and
+                            member_data.get('version') == data.get('version') and
+                            member_data.get('checkpoint_after_promote') == data.get('checkpoint_after_promote'))):
            try:
                self._client.delete_async(self.member_path).get(timeout=1)
            except NoNodeError:
@@ -347,13 +365,15 @@ class ZooKeeper(AbstractDCS):
                return False
            member = None

+        encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8')
        if member:
-            if deep_compare(data, member.data):
+            if deep_compare(data, member_data):
                return True
        else:
            try:
                self._client.create_async(self.member_path, encoded_data, makepath=True,
                                          ephemeral=not permanent).get(timeout=1)
+                self.__last_member_data = data
                return True
            except Exception as e:
                if not isinstance(e, NodeExistsError):
@@ -361,6 +381,7 @@ class ZooKeeper(AbstractDCS):
                    return False
        try:
            self._client.set_async(self.member_path, encoded_data).get(timeout=1)
+            self.__last_member_data = data
            return True
        except Exception:
            logger.exception('touch_member')
@@ -2,32 +2,36 @@ import datetime
 import functools
 import json
 import logging
-import psycopg2
+import six
 import sys
 import time
 import uuid

 from collections import namedtuple
 from multiprocessing.pool import ThreadPool
-from patroni.async_executor import AsyncExecutor, CriticalTask
-from patroni.exceptions import DCSError, PostgresConnectionException, PatroniFatalException
-from patroni.postgresql import ACTION_ON_START, ACTION_ON_ROLE_CHANGE
-from patroni.postgresql.misc import postgres_version_to_int
-from patroni.postgresql.rewind import Rewind
-from patroni.utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster, parse_int
-from patroni.dcs import RemoteMember
 from threading import RLock

+from . import psycopg
+from .async_executor import AsyncExecutor, CriticalTask
+from .exceptions import DCSError, PostgresConnectionException, PatroniFatalException
+from .postgresql import ACTION_ON_START, ACTION_ON_ROLE_CHANGE
+from .postgresql.misc import postgres_version_to_int
+from .postgresql.rewind import Rewind
+from .utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster, parse_int
+from .dcs import RemoteMember
+
 logger = logging.getLogger(__name__)


-class _MemberStatus(namedtuple('_MemberStatus', ['member', 'reachable', 'in_recovery', 'timeline',
-                                                 'wal_position', 'tags', 'watchdog_failed'])):
+class _MemberStatus(namedtuple('_MemberStatus', ['member', 'reachable', 'in_recovery',
+                                                 'dcs_last_seen', 'timeline', 'wal_position',
+                                                 'tags', 'watchdog_failed'])):
    """Node status distilled from API response:

        member - dcs.Member object of the node
        reachable - `!False` if the node is not reachable or is not responding with correct JSON
        in_recovery - `!True` if pg_is_in_recovery() == true
+        dcs_last_seen - timestamp from JSON of last succesful communication with DCS
        timeline - timeline value from JSON
        wal_position - maximum value of `replayed_location` or `received_location` from JSON
        tags - dictionary with values of different tags (i.e. nofailover)
@@ -37,12 +41,14 @@ class _MemberStatus(namedtuple('_MemberStatus', ['member', 'reachable', 'in_reco
    def from_api_response(cls, member, json):
        is_master = json['role'] == 'master'
        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))
-        return cls(member, True, not is_master, timeline, wal, json.get('tags', {}), json.get('watchdog_failed', False))
+        return cls(member, True, not is_master, dcs_last_seen, timeline, wal,
+                   json.get('tags', {}), json.get('watchdog_failed', False))

    @classmethod
    def unknown(cls, member):
-        return cls(member, False, None, 0, 0, {}, False)
+        return cls(member, False, None, 0, 0, 0, {}, False)

    def failover_limitation(self):
        """Returns reason why this node can't promote or None if everything is ok."""
@@ -186,9 +192,6 @@ class Ha(object):
                'version': self.patroni.version
            }

-            # following two lines are mainly necessary for consul, to avoid creation of master service
-            if data['role'] == 'master' and not self.is_leader():
-                data['role'] = 'promoted'
            if self.is_leader() and not self._rewind.checkpoint_after_promote():
                data['checkpoint_after_promote'] = False
            tags = self.get_effective_tags()
@@ -313,10 +316,7 @@ class Ha(object):
            if timeout == 0:
                # We are requested to prefer failing over to restarting master. But see first if there
                # is anyone to fail over to.
-                members = self.cluster.members
-                if self.is_synchronous_mode():
-                    members = [m for m in members if self.cluster.sync.matches(m.name)]
-                if self.is_failover_possible(members):
+                if self.is_failover_possible(self.cluster.members):
                    logger.info("Master crashed. Failing over.")
                    self.demote('immediate')
                    return 'stopped PostgreSQL to fail over after a crash'
@@ -686,16 +686,21 @@ class Ha(object):
                        logger.info('Ignoring the former leader being ahead of us')
        return True

-    def is_failover_possible(self, members):
+    def is_failover_possible(self, members, check_synchronous=True, cluster_lsn=None):
        ret = False
        cluster_timeline = self.cluster.timeline
        members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url]
+        if check_synchronous and self.is_synchronous_mode():
+            members = [m for m in members if self.cluster.sync.matches(m.name)]
        if members:
            for st in self.fetch_nodes_statuses(members):
                not_allowed_reason = st.failover_limitation()
                if not_allowed_reason:
                    logger.info('Member %s is %s', st.member.name, not_allowed_reason)
-                elif self.is_lagging(st.wal_position):
+                elif not isinstance(st.wal_position, six.integer_types):
+                    logger.info('Member %s does not report wal_position', st.member.name)
+                elif cluster_lsn and st.wal_position < cluster_lsn or\
+                        not cluster_lsn and self.is_lagging(st.wal_position):
                    logger.info('Member %s exceeds maximum replication lag', st.member.name)
                elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline):
                    logger.info('Timeline %s of member %s is behind the cluster timeline %s',
@@ -779,6 +784,10 @@ class Ha(object):
            return False

        if self.cluster.failover:
+            # When doing a switchover in synchronous mode only synchronous nodes and former leader are allowed to race
+            if self.is_synchronous_mode() and self.cluster.failover.leader and \
+                    self.cluster.failover.candidate and not self.cluster.sync.matches(self.state_handler.name):
+                return False
            return self.manual_failover_process_no_leader()

        if not self.watchdog.is_healthy:
@@ -826,23 +835,44 @@ class Ha(object):
            'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async_req=True),
        }[mode]

+        logger.info('Demoting self (%s)', mode)
+
        self._rewind.trigger_check_diverged_lsn()
+
+        status = {'released': False}
+
+        def on_shutdown(checkpoint_location):
+            # Postmaster is still running, but pg_control already reports clean "shut down".
+            # It could happen if Postgres is still archiving the backlog of WAL files.
+            # If we know that there are replicas that received the shutdown checkpoint
+            # location, we can remove the leader key and allow them to start leader race.
+            if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location):
+                self.state_handler.set_role('demoted')
+                with self._async_executor:
+                    self.release_leader_key_voluntarily(checkpoint_location)
+                    status['released'] = True
+
        self.state_handler.stop(mode_control['stop'], checkpoint=mode_control['checkpoint'],
                                on_safepoint=self.watchdog.disable if self.watchdog.is_running else None,
+                                on_shutdown=on_shutdown if mode_control['release'] else None,
                                stop_timeout=self.master_stop_timeout())
        self.state_handler.set_role('demoted')
        self.set_is_leader(False)

        if mode_control['release']:
-            checkpoint_location = self.state_handler.latest_checkpoint_location() if mode == 'graceful' else None
-            with self._async_executor:
-                self.release_leader_key_voluntarily(checkpoint_location)
+            if not status['released']:
+                checkpoint_location = self.state_handler.latest_checkpoint_location() if mode == 'graceful' else None
+                with self._async_executor:
+                    self.release_leader_key_voluntarily(checkpoint_location)
            time.sleep(2)  # Give a time to somebody to take the leader lock
        if mode_control['offline']:
            node_to_follow, leader = None, None
        else:
-            cluster = self.dcs.get_cluster()
-            node_to_follow, leader = self._get_node_to_follow(cluster), cluster.leader
+            try:
+                cluster = self.dcs.get_cluster()
+                node_to_follow, leader = self._get_node_to_follow(cluster), cluster.leader
+            except Exception:
+                node_to_follow, leader = None, None

        # FIXME: with mode offline called from DCS exception handler and handle_long_action_in_progress
        # there could be an async action already running, calling follow from here will lead
@@ -920,7 +950,7 @@ class Ha(object):
                    else:
                        members = [m for m in self.cluster.members
                                   if not failover.candidate or m.name == failover.candidate]
-                    if self.is_failover_possible(members):  # check that there are healthy members
+                    if self.is_failover_possible(members, False):  # check that there are healthy members
                        ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',))
                        return ret or 'manual failover: demoting myself'
                    else:
@@ -997,13 +1027,13 @@ class Ha(object):
                    # in case of standby cluster we don't really need to
                    # enforce anything, since the leader is not a master.
                    # So just remind the role.
-                    msg = 'no action. I am ({0}) the standby leader with the lock'.format(self.state_handler.name) \
+                    msg = 'no action. I am ({0}), the standby leader with the lock'.format(self.state_handler.name) \
                          if self.state_handler.role == 'standby_leader' else \
                          'promoted self to a standby leader because i had the session lock'
                    return self.enforce_follow_remote_master(msg)
                else:
                    return self.enforce_master_role(
-                        'no action. I am ({0}) the leader with the lock'.format(self.state_handler.name),
+                        'no action. I am ({0}), the leader with the lock'.format(self.state_handler.name),
                        'promoted self to leader because I had the session lock'
                    )
            else:
@@ -1021,10 +1051,10 @@ class Ha(object):
        lock_owner = self.cluster.leader and self.cluster.leader.name
        if self.is_standby_cluster():
            return self.follow('cannot be a real primary in a standby cluster',
-                               'no action. I am a secondary ({0}) and following a standby leader ({1})'.format(
+                               'no action. I am ({0}), a secondary, and following a standby leader ({1})'.format(
                                    self.state_handler.name, lock_owner), refresh=False)
        return self.follow('demoting self because I do not have the lock and I was a leader',
-                           'no action. I am a secondary ({0}) and following a leader ({1})'.format(
+                           'no action. I am ({0}), a secondary, and following a leader ({1})'.format(
                                self.state_handler.name, lock_owner), refresh=False)

    def evaluate_scheduled_restart(self):
@@ -1250,6 +1280,7 @@ class Ha(object):
        if not self.watchdog.activate():
            logger.error('Cancelling bootstrap because watchdog activation failed')
            self.cancel_initialization()
+        self._rewind.ensure_checkpoint_after_promote(self.wakeup)
        self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid)
        self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
        self.dcs.take_leader()
@@ -1459,7 +1490,7 @@ class Ha(object):
                self.demote('offline')
                return 'demoted self because DCS is not accessible and i was a leader'
            return 'DCS is not accessible'
-        except (psycopg2.Error, PostgresConnectionException):
+        except (psycopg.Error, PostgresConnectionException):
            return 'Error communicating with PostgreSQL. Will try again later'
        finally:
            if not dcs_failed:
@@ -1486,10 +1517,27 @@ class Ha(object):
            # This might not be the desired behavior of users, as a graceful shutdown of the host can mean lost data.
            # We probably need to something smarter here.
            disable_wd = self.watchdog.disable if self.watchdog.is_running else None
+
+            status = {'deleted': False}
+
+            def _on_shutdown(checkpoint_location):
+                if self.is_leader():
+                    # Postmaster is still running, but pg_control already reports clean "shut down".
+                    # It could happen if Postgres is still archiving the backlog of WAL files.
+                    # If we know that there are replicas that received the shutdown checkpoint
+                    # location, we can remove the leader key and allow them to start leader race.
+                    if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location):
+                        self.dcs.delete_leader(checkpoint_location)
+                        status['deleted'] = True
+                    else:
+                        self.dcs.write_leader_optime(checkpoint_location)
+
+            on_shutdown = _on_shutdown if self.is_leader() else None
            self.while_not_sync_standby(lambda: self.state_handler.stop(checkpoint=False, on_safepoint=disable_wd,
+                                                                        on_shutdown=on_shutdown,
                                                                        stop_timeout=self.master_stop_timeout()))
            if not self.state_handler.is_running():
-                if self.is_leader():
+                if self.is_leader() and not status['deleted']:
                    checkpoint_location = self.state_handler.latest_checkpoint_location()
                    self.dcs.delete_leader(checkpoint_location)
                self.touch_member()
@@ -1,6 +1,5 @@
 import logging
 import os
-import psycopg2
 import re
 import shlex
 import shutil
@@ -23,6 +22,7 @@ from .connection import Connection, get_connection_cursor
 from .misc import parse_history, parse_lsn, postgres_major_version_to_int
 from .postmaster import PostmasterProcess
 from .slots import SlotsHandler
+from .. import psycopg
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int

@@ -266,13 +266,13 @@ class Postgresql(object):
            cursor = self._connection.cursor()
            cursor.execute(sql, params)
            return cursor
-        except psycopg2.Error as e:
+        except psycopg.Error as e:
            if cursor and cursor.connection.closed == 0:
                # When connected via unix socket, psycopg2 can't recoginze 'connection lost'
                # and leaves `_cursor_holder.connection.closed == 0`, but psycopg2.OperationalError
                # is still raised (what is correct). It doesn't make sense to continiue with existing
                # connection and we will close it, to avoid its reuse by the `cursor` method.
-                if isinstance(e, psycopg2.OperationalError):
+                if isinstance(e, psycopg.OperationalError):
                    self._connection.close()
                else:
                    raise e
@@ -327,6 +327,9 @@ class Postgresql(object):
        if cluster and cluster.config and cluster.config.modify_index:
            self._has_permanent_logical_slots =\
                cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version)
+
+            # We want to enable hot_standby_feedback if the replica is supposed
+            # to have a logical slot or in case if it is the cascading replica.
            self.set_enforce_hot_standby_feedback(
                self._has_permanent_logical_slots or
                cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
@@ -338,7 +341,9 @@ class Postgresql(object):
                cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location',
                                               'received_location', 'replay_paused', 'pg_control_timeline',
                                               'received_tli', 'slot_name', 'conninfo', 'slots'], result))
-                cluster_info_state['slots'] = self.slots_handler.process_permanent_slots(cluster_info_state['slots'])
+                if self._has_permanent_logical_slots:
+                    cluster_info_state['slots'] =\
+                        self.slots_handler.process_permanent_slots(cluster_info_state['slots'])
                self._cluster_info_state = cluster_info_state
            except RetryFailedError as e:  # SELECT failed two times
                self._cluster_info_state = {'error': str(e)}
@@ -369,7 +374,11 @@ class Postgresql(object):
        return self._cluster_info_state_get('received_tli')

    def is_leader(self):
-        return bool(self._cluster_info_state_get('timeline'))
+        try:
+            return bool(self._cluster_info_state_get('timeline'))
+        except PostgresConnectionException:
+            logger.warning('Failed to determine PostgreSQL state from the connection, falling back to cached role')
+            return bool(self.is_running() and self.role == 'master')

    def replay_paused(self):
        return self._cluster_info_state_get('replay_paused')
@@ -384,6 +393,7 @@ class Postgresql(object):

    def pg_control_timeline(self):
        try:
+
            return int(self.controldata().get("Latest checkpoint's TimeLineID"))
        except (TypeError, ValueError):
            logger.exception('Failed to parse timeline from pg_controldata output')
@@ -593,12 +603,13 @@ class Postgresql(object):
                    cur.execute('SELECT pg_catalog.pg_is_in_recovery()')
                    if cur.fetchone()[0]:
                        return 'is_in_recovery=true'
-                return cur.execute('CHECKPOINT')
-        except psycopg2.Error:
+                cur.execute('CHECKPOINT')
+        except psycopg.Error:
            logger.exception('Exception during CHECKPOINT')
            return 'not accessible or not healty'

-    def stop(self, mode='fast', block_callbacks=False, checkpoint=None, on_safepoint=None, stop_timeout=None):
+    def stop(self, mode='fast', block_callbacks=False, checkpoint=None,
+             on_safepoint=None, on_shutdown=None, stop_timeout=None):
        """Stop PostgreSQL

        Supports a callback when a safepoint is reached. A safepoint is when no user backend can return a successful
@@ -606,11 +617,12 @@ class Postgresql(object):
        could be added.

        :param on_safepoint: This callback is called when no user backends are running.
+        :param on_shutdown: is called when pg_controldata starts reporting `Database cluster state: shut down`
        """
        if checkpoint is None:
            checkpoint = False if mode == 'immediate' else True

-        success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint, on_safepoint, stop_timeout)
+        success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint, on_safepoint, on_shutdown, stop_timeout)
        if success:
            # block_callbacks is used during restart to avoid
            # running start/stop callbacks in addition to restart ones
@@ -623,7 +635,7 @@ class Postgresql(object):
            self.set_state('stop failed')
        return success

-    def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint, stop_timeout):
+    def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint, on_shutdown, stop_timeout):
        postmaster = self.is_running()
        if not postmaster:
            if on_safepoint:
@@ -650,6 +662,22 @@ class Postgresql(object):
            postmaster.wait_for_user_backends_to_close()
            on_safepoint()

+        if on_shutdown and mode in ('fast', 'smart'):
+            i = 0
+            # Wait for pg_controldata `Database cluster state:` to change to "shut down"
+            while postmaster.is_running():
+                data = self.controldata()
+                if data.get('Database cluster state', '') == 'shut down':
+                    on_shutdown(int(self.latest_checkpoint_location()))
+                    break
+                elif data.get('Database cluster state', '').startswith('shut down'):  # shut down in recovery
+                    break
+                elif stop_timeout and i >= stop_timeout:
+                    stop_timeout = 0
+                    break
+                time.sleep(STOP_POLLING_INTERVAL)
+                i += STOP_POLLING_INTERVAL
+
        try:
            postmaster.wait(timeout=stop_timeout)
        except TimeoutExpired:
@@ -684,7 +712,7 @@ class Postgresql(object):
                while postmaster.is_running():  # Need a timeout here?
                    cur.execute("SELECT 1")
                    time.sleep(STOP_POLLING_INTERVAL)
-        except psycopg2.Error:
+        except psycopg.Error:
            pass

    def reload(self, block_callbacks=False):
@@ -809,7 +837,7 @@ class Postgresql(object):
            return None, None

    @contextmanager
-    def get_replication_connection_cursor(self, host='localhost', port=5432, **kwargs):
+    def get_replication_connection_cursor(self, host=None, port=5432, **kwargs):
        conn_kwargs = self.config.replication.copy()
        conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
                           connect_timeout=3, replication=1, options='-c statement_timeout=2000')
@@ -861,20 +889,22 @@ class Postgresql(object):
        if change_role:
            self.__cb_pending = ACTION_NOOP

+        ret = True
        if self.is_running():
            if do_reload:
                self.config.write_postgresql_conf()
-                if self.reload(block_callbacks=change_role) and change_role:
+                ret = self.reload(block_callbacks=change_role)
+                if ret and change_role:
                    self.set_role(role)
            else:
-                self.restart(block_callbacks=change_role, role=role)
+                ret = self.restart(block_callbacks=change_role, role=role)
        else:
-            self.start(timeout=timeout, block_callbacks=change_role, role=role)
+            ret = self.start(timeout=timeout, block_callbacks=change_role, role=role) or None

        if change_role:
            # TODO: postpone this until start completes, or maybe do even earlier
            self.call_nowait(ACTION_ON_ROLE_CHANGE)
-        return True
+        return ret

    def _wait_promote(self, wait_seconds):
        for _ in polling_loop(wait_seconds):
@@ -955,7 +985,7 @@ class Postgresql(object):
            with self.connection().cursor() as cursor:
                cursor.execute(query)
                return cursor.fetchone()[0].isoformat(sep=' ')
-        except psycopg2.Error:
+        except psycopg.Error:
            return None

    def last_operation(self):
@@ -1082,11 +1112,12 @@ class Postgresql(object):
                " ORDER BY sync_state DESC, {0}_{1} DESC".format(sort_col, self.lsn_name, self.wal_name)):
            member = members.get(app_name)
            if member and not member.tags.get('nosync', False):
-                replica_list.append((member.name, sync_state, replica_lsn))
+                replica_list.append((member.name, sync_state, replica_lsn, bool(member.nofailover)))

        max_lsn = max(replica_list, key=lambda x: x[2])[2] if len(replica_list) > 1 else int(str(self.last_operation()))

-        for app_name, sync_state, replica_lsn in replica_list:
+        # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
+        for app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[3]):
            if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
                candidates.append(app_name)
                if sync_state == 'sync':
@@ -4,10 +4,12 @@ import shlex
 import tempfile
 import time

-from patroni.dcs import RemoteMember
-from patroni.utils import deep_compare
 from six import string_types

+from ..dcs import RemoteMember
+from ..psycopg import quote_ident, quote_literal
+from ..utils import deep_compare
+
 logger = logging.getLogger(__name__)


@@ -297,26 +299,24 @@ class Bootstrap(object):
        if 'NOLOGIN' not in options and 'LOGIN' not in options:
            options.append('LOGIN')

-        params = [name]
        if password:
-            options.extend(['PASSWORD', '%s'])
-            params.extend([password, password])
+            options.extend(['PASSWORD', quote_literal(password)])

        sql = """DO $$
 BEGIN
    SET local synchronous_commit = 'local';
-    PERFORM * FROM pg_authid WHERE rolname = %s;
+    PERFORM * FROM pg_catalog.pg_authid WHERE rolname = {0};
    IF FOUND THEN
-        ALTER ROLE "{0}" WITH {1};
+        ALTER ROLE {1} WITH {2};
    ELSE
-        CREATE ROLE "{0}" WITH {1};
+        CREATE ROLE {1} WITH {2};
    END IF;
-END;$$""".format(name, ' '.join(options))
+END;$$""".format(quote_literal(name), quote_ident(name, self._postgresql.connection()), ' '.join(options))
        self._postgresql.query('SET log_statement TO none')
        self._postgresql.query('SET log_min_duration_statement TO -1')
        self._postgresql.query("SET log_min_error_statement TO 'log'")
        try:
-            self._postgresql.query(sql, *params)
+            self._postgresql.query(sql)
        finally:
            self._postgresql.query('RESET log_min_error_statement')
            self._postgresql.query('RESET log_min_duration_statement')
@@ -342,8 +342,8 @@ END;$$""".format(name, ' '.join(options))
                        sql = """DO $$
 BEGIN
    SET local synchronous_commit = 'local';
-    GRANT EXECUTE ON function pg_catalog.{0} TO "{1}";
-END;$$""".format(f, rewind['username'])
+    GRANT EXECUTE ON function pg_catalog.{0} TO {1};
+END;$$""".format(f, quote_ident(rewind['username'], self._postgresql.connection()))
                        postgresql.query(sql)

                for name, value in (config.get('users') or {}).items():
@@ -12,6 +12,7 @@ from .validator import CaseInsensitiveDict, recovery_parameters,\
        transform_postgresql_parameter_value, transform_recovery_parameter_value
 from ..dcs import slot_name_from_member_name, RemoteMember
 from ..exceptions import PatroniFatalException
+from ..psycopg import quote_ident as _quote_ident
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, \
        validate_directory, is_subpath

@@ -23,7 +24,7 @@ PARAMETER_RE = re.compile(r'([a-z_]+)\s*=\s*')

 def quote_ident(value):
    """Very simplified version of quote_ident"""
-    return value if SYNC_STANDBY_NAME_RE.match(value) else '"' + value + '"'
+    return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value)


 def conninfo_uri_parse(dsn):
@@ -477,18 +478,19 @@ class ConfigHandler(object):
            ret.setdefault('channel_binding', 'prefer')
        if self._krbsrvname:
            ret['krbsrvname'] = self._krbsrvname
-        if 'database' in ret:
-            del ret['database']
+        if 'dbname' in ret:
+            del ret['dbname']
        return ret

    def format_dsn(self, params, include_dbname=False):
        # A list of keywords that can be found in a conninfo string. Follows what is acceptable by libpq
        keywords = ('dbname', 'user', 'passfile' if params.get('passfile') else 'password', 'host', 'port',
                    'sslmode', 'sslcompression', 'sslcert', 'sslkey', 'sslpassword', 'sslrootcert', 'sslcrl',
-                    'application_name', 'krbsrvname', 'gssencmode', 'channel_binding')
+                    'sslcrldir', 'application_name', 'krbsrvname', 'gssencmode', 'channel_binding')
        if include_dbname:
            params = params.copy()
-            params['dbname'] = params.get('database') or self._postgresql.database
+            if 'dbname' not in params:
+                params['dbname'] = self._postgresql.database
            # we are abusing information about the necessity of dbname
            # dsn should contain passfile or password only if there is no dbname in it (it is used in recovery.conf)
            skip = {'passfile', 'password'}
@@ -834,7 +836,7 @@ class ConfigHandler(object):
        # this exercise is improving cross version compatibility and user must set the correct parameter in the config.
        if self._postgresql.major_version >= 130000:
            wal_keep_segments = parameters.pop('wal_keep_segments', self.CMDLINE_OPTIONS['wal_keep_segments'][0])
-            parameters.setdefault('wal_keep_size', str(wal_keep_segments * 16) + 'MB')
+            parameters.setdefault('wal_keep_size', str(int(wal_keep_segments) * 16) + 'MB')
        elif self._postgresql.major_version:
            wal_keep_size = parse_int(parameters.pop('wal_keep_size', self.CMDLINE_OPTIONS['wal_keep_size'][0]), 'MB')
            parameters.setdefault('wal_keep_segments', int((wal_keep_size + 8) / 16))
@@ -870,7 +872,7 @@ class ConfigHandler(object):
            ret['user'] = self._superuser['username']
            del ret['username']
        # ensure certain Patroni configurations are available
-        ret.update({'database': self._postgresql.database,
+        ret.update({'dbname': self._postgresql.database,
                    'fallback_application_name': 'Patroni',
                    'connect_timeout': 3,
                    'options': '-c statement_timeout=2000'})
@@ -1,9 +1,10 @@
 import logging
-import psycopg2

 from contextlib import contextmanager
 from threading import Lock

+from .. import psycopg
+
 logger = logging.getLogger(__name__)


@@ -20,7 +21,7 @@ class Connection(object):
    def get(self):
        with self._lock:
            if not self._connection or self._connection.closed != 0:
-                self._connection = psycopg2.connect(**self._conn_kwargs)
+                self._connection = psycopg.connect(**self._conn_kwargs)
                self._connection.autocommit = True
                self.server_version = self._connection.server_version
        return self._connection
@@ -40,7 +41,7 @@ class Connection(object):

@contextmanager
 def get_connection_cursor(**kwargs):
-    conn = psycopg2.connect(**kwargs)
+    conn = psycopg.connect(**kwargs)
    conn.autocommit = True
    with conn.cursor() as cur:
        yield cur
@@ -152,8 +152,8 @@ class Rewind(object):

    def _conn_kwargs(self, member, auth):
        ret = member.conn_kwargs(auth)
-        if not ret.get('database'):
-            ret['database'] = self._postgresql.database
+        if not ret.get('dbname'):
+            ret['dbname'] = self._postgresql.database
        return ret

    def _check_timeline_and_lsn(self, leader):
@@ -161,11 +161,11 @@ class Rewind(object):
        if local_timeline is None or local_lsn is None:
            return

-        if isinstance(leader, Leader):
-            if leader.member.data.get('role') != 'master':
-                return
-        # standby cluster
-        elif not self.check_leader_is_not_in_recovery(self._conn_kwargs(leader, self._postgresql.config.replication)):
+        if isinstance(leader, Leader) and leader.member.data.get('role') != 'master':
+            return
+
+        if not self.check_leader_is_not_in_recovery(
+                self._conn_kwargs(leader, self._postgresql.config.rewind_credentials)):
            return

        history = need_rewind = None
@@ -179,7 +179,7 @@ class Rewind(object):
                elif local_timeline == master_timeline:
                    need_rewind = False
                elif master_timeline > 1:
-                    cur.execute('TIMELINE_HISTORY %s', (master_timeline,))
+                    cur.execute('TIMELINE_HISTORY {0}'.format(master_timeline))
                    history = cur.fetchone()[1]
                    if not isinstance(history, six.string_types):
                        history = bytes(history).decode('utf-8')
@@ -200,9 +200,11 @@ class Rewind(object):
                        need_rewind = True
                    else:
                        need_rewind = switchpoint != self._get_checkpoint_end(local_timeline, local_lsn)
-                    break
                elif parent_timeline > local_timeline:
+                    need_rewind = True
                    break
+            else:
+                need_rewind = True
            self._log_master_history(history, i)

        self._state = need_rewind and REWIND_STATUS.NEED or REWIND_STATUS.NOT_NEED
@@ -230,16 +232,14 @@ class Rewind(object):
            with self._checkpoint_task_lock:
                if self._checkpoint_task:
                    with self._checkpoint_task:
-                        if self._checkpoint_task.result:
+                        if self._checkpoint_task.result is not None:
                            self._state = REWIND_STATUS.CHECKPOINT
-                        if self._checkpoint_task.result is not False:
-                            return
+                            self._checkpoint_task = None
+                elif self._postgresql.get_master_timeline() == self._postgresql.pg_control_timeline():
+                    self._state = REWIND_STATUS.CHECKPOINT
                else:
                    self._checkpoint_task = CriticalTask()
-                    return Thread(target=self.__checkpoint, args=(self._checkpoint_task, wakeup)).start()
-
-            if self._postgresql.get_master_timeline() == self._postgresql.pg_control_timeline():
-                self._state = REWIND_STATUS.CHECKPOINT
+                    Thread(target=self.__checkpoint, args=(self._checkpoint_task, wakeup)).start()

    def checkpoint_after_promote(self):
        return self._state == REWIND_STATUS.CHECKPOINT
@@ -5,10 +5,10 @@ import shutil

 from collections import defaultdict
 from contextlib import contextmanager
-from psycopg2.errors import UndefinedFile

 from .connection import get_connection_cursor
 from .misc import format_lsn
+from ..psycopg import UndefinedFile

 logger = logging.getLogger(__name__)

@@ -82,8 +82,9 @@ class SlotsHandler(object):
            replication_slots = {}
            extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"\
                if self._postgresql.major_version >= 100000 else ""
+            skip_temp_slots = ' WHERE NOT temporary' if self._postgresql.major_version >= 100000 else ''
            cursor = self._query('SELECT slot_name, slot_type, plugin, database, datoid'
-                                 '{0} FROM pg_catalog.pg_replication_slots'.format(extra))
+                                 '{0} FROM pg_catalog.pg_replication_slots{1}'.format(extra, skip_temp_slots))
            for r in cursor:
                value = {'type': r[1]}
                if r[1] == 'logical':
@@ -142,9 +143,9 @@ class SlotsHandler(object):
                self._schedule_load_slots = True

    @contextmanager
-    def _get_local_connection_cursor(self, database):
+    def _get_local_connection_cursor(self, **kwargs):
        conn_kwargs = self._postgresql.config.local_connect_kwargs
-        conn_kwargs['database'] = database
+        conn_kwargs.update(kwargs)
        with get_connection_cursor(**conn_kwargs) as cur:
            yield cur

@@ -161,7 +162,7 @@ class SlotsHandler(object):

        # Create new logical slots
        for database, values in logical_slots.items():
-            with self._get_local_connection_cursor(database) as cur:
+            with self._get_local_connection_cursor(dbname=database) as cur:
                for name, value in values.items():
                    try:
                        cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" +
@@ -193,7 +194,7 @@ class SlotsHandler(object):

        # Advance logical slots
        for database, values in advance_slots.items():
-            with self._get_local_connection_cursor(database) as cur:
+            with self._get_local_connection_cursor(dbname=database, options='-c statement_timeout=0') as cur:
                for name, value in values.items():
                    try:
                        cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)",
@@ -235,7 +236,7 @@ class SlotsHandler(object):
    @contextmanager
    def _get_leader_connection_cursor(self, leader):
        conn_kwargs = leader.conn_kwargs(self._postgresql.config.rewind_credentials)
-        conn_kwargs['database'] = self._postgresql.database
+        conn_kwargs['dbname'] = self._postgresql.database
        with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
            yield cur

@@ -0,0 +1,43 @@
+__all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError',
+           'Error', 'OperationalError', 'ProgrammingError', 'UndefinedFile']
+
+_legacy = False
+try:
+    from psycopg2 import __version__
+    from . import MIN_PSYCOPG2, parse_version
+    if parse_version(__version__) < MIN_PSYCOPG2:
+        raise ImportError
+    from psycopg2 import connect, Error, DatabaseError, OperationalError, ProgrammingError
+    from psycopg2.errors import UndefinedFile
+    from psycopg2.extensions import adapt
+
+    try:
+        from psycopg2.extensions import quote_ident as _quote_ident
+    except ImportError:
+        _legacy = True
+
+    def quote_literal(value, conn=None):
+        value = adapt(value)
+        if conn:
+            value.prepare(conn)
+        return value.getquoted().decode('utf-8')
+except ImportError:
+    from psycopg import connect as _connect, sql, Error, DatabaseError, OperationalError, ProgrammingError
+    from psycopg.errors import UndefinedFile
+
+    def connect(*args, **kwargs):
+        ret = _connect(*args, **kwargs)
+        ret.server_version = ret.pgconn.server_version  # compatibility with psycopg2
+        return ret
+
+    def _quote_ident(value, conn):
+        return sql.Identifier(value).as_string(conn)
+
+    def quote_literal(value, conn=None):
+        return sql.Literal(value).as_string(conn)
+
+
+def quote_ident(value, conn=None):
+    if _legacy or conn is None:
+        return '"{0}"'.format(value.replace('"', '""'))
+    return _quote_ident(value, conn)
@@ -27,13 +27,14 @@ import argparse
 import csv
 import logging
 import os
-import psycopg2
 import subprocess
 import sys
 import time

 from collections import namedtuple

+from .. import psycopg
+
 logger = logging.getLogger(__name__)

 RETRY_SLEEP_INTERVAL = 1
@@ -215,7 +216,7 @@ class WALERestore(object):
            if self.master_connection:
                try:
                    # get the difference in bytes between the current WAL location and the backup start offset
-                    with psycopg2.connect(self.master_connection) as con:
+                    with psycopg.connect(self.master_connection) as con:
                        if con.server_version >= 100000:
                            wal_name = 'wal'
                            lsn_name = 'lsn'
@@ -233,7 +234,7 @@ class WALERestore(object):
                                        (backup_start_lsn, backup_start_lsn, backup_start_lsn))

                            diff_in_bytes = int(cur.fetchone()[0])
-                except psycopg2.Error:
+                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
                        attempts_no = attempts_no + 1
@@ -1 +1 @@
-__version__ = '2.1.1'
+__version__ = '2.1.2'
@@ -216,13 +216,13 @@ def setup_package(version):
 if __name__ == '__main__':
    old_modules = sys.modules.copy()
    try:
-        from patroni import check_psycopg2, fatal, __version__
+        from patroni import check_psycopg, fatal, __version__
    finally:
        sys.modules.clear()
        sys.modules.update(old_modules)

    if sys.version_info < (2, 7, 0):
        fatal('Patroni needs to be run with Python 2.7+')
-    check_psycopg2()
+    check_psycopg()

    setup_package(__version__)
@@ -5,9 +5,10 @@ import unittest

 from mock import Mock, patch

-import psycopg2
 import urllib3

+import patroni.psycopg as psycopg
+
 from patroni.dcs import Leader, Member
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
@@ -85,9 +86,9 @@ class MockCursor(object):

    def execute(self, sql, *params):
        if sql.startswith('blabla'):
-            raise psycopg2.ProgrammingError()
+            raise psycopg.ProgrammingError()
        elif sql == 'CHECKPOINT' or sql.startswith('SELECT pg_catalog.pg_create_'):
-            raise psycopg2.OperationalError()
+            raise psycopg.OperationalError()
        elif sql.startswith('RetryFailedError'):
            raise RetryFailedError('retry')
        elif sql.startswith('SELECT catalog_xmin'):
@@ -162,7 +163,7 @@ class MockConnect(object):
        pass


-def psycopg2_connect(*args, **kwargs):
+def psycopg_connect(*args, **kwargs):
    return MockConnect()


@@ -176,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('psycopg2.connect', psycopg2_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())
@@ -1,9 +1,10 @@
 import datetime
 import json
-import psycopg2
 import unittest
 import socket

+import patroni.psycopg as psycopg
+
 from mock import Mock, PropertyMock, patch
 from patroni.api import RestApiHandler, RestApiServer
 from patroni.dcs import ClusterConfig, Member
@@ -11,7 +12,7 @@ from patroni.ha import _MemberStatus
 from patroni.utils import tzutc
 from six import BytesIO as IO
 from six.moves import BaseHTTPServer
-from . import psycopg2_connect, MockCursor
+from . import psycopg_connect, MockCursor
 from .test_ha import get_cluster_initialized_without_leader


@@ -35,7 +36,7 @@ class MockPostgresql(object):

    @staticmethod
    def connection():
-        return psycopg2_connect()
+        return psycopg_connect()

    @staticmethod
    def postmaster_start_time():
@@ -77,7 +78,7 @@ class MockHa(object):

    @staticmethod
    def fetch_nodes_statuses(members):
-        return [_MemberStatus(None, True, None, 0, None, {}, False)]
+        return [_MemberStatus(None, True, None, 0, 0, None, {}, False)]

    @staticmethod
    def schedule_future_restart(data):
@@ -435,9 +436,9 @@ class TestRestApiHandler(unittest.TestCase):

    @patch('time.sleep', Mock())
    def test_RestApiServer_query(self):
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)):
            self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
-        with patch.object(MockPostgresql, 'connection', Mock(side_effect=psycopg2.OperationalError)):
+        with patch.object(MockPostgresql, 'connection', Mock(side_effect=psycopg.OperationalError)):
            self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))

    @patch('time.sleep', Mock())
@@ -8,11 +8,11 @@ from patroni.postgresql.bootstrap import Bootstrap
 from patroni.postgresql.cancellable import CancellableSubprocess
 from patroni.postgresql.config import ConfigHandler

-from . import psycopg2_connect, BaseTestPostgresql
+from . import psycopg_connect, BaseTestPostgresql


@patch('subprocess.call', Mock(return_value=0))
-@patch('psycopg2.connect', psycopg2_connect)
+@patch('patroni.psycopg.connect', psycopg_connect)
@patch('os.rename', Mock())
 class TestBootstrap(BaseTestPostgresql):

@@ -164,6 +164,7 @@ class TestBootstrap(BaseTestPostgresql):
    @patch('os.unlink', Mock())
    @patch('shutil.copy', Mock())
    @patch('os.path.isfile', Mock(return_value=True))
+    @patch('patroni.postgresql.bootstrap.quote_ident', Mock())
    @patch.object(Bootstrap, 'call_post_bootstrap', Mock(return_value=True))
    @patch.object(Bootstrap, '_custom_bootstrap', Mock(return_value=True))
    @patch.object(Postgresql, 'start', Mock(return_value=True))
@@ -2,7 +2,7 @@ import consul
 import unittest

 from consul import ConsulException, NotFound
-from mock import Mock, patch
+from mock import Mock, PropertyMock, patch
 from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \
                                ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession
 from . import SleepException
@@ -154,8 +154,10 @@ class TestConsul(unittest.TestCase):
    def test_set_config_value(self):
        self.c.set_config_value('')

+    @patch.object(Cluster, 'min_version', PropertyMock(return_value=(2, 0)))
    @patch.object(consul.Consul.KV, 'put', Mock(side_effect=ConsulException))
    def test_write_leader_optime(self):
+        self.c.get_cluster()
        self.c.write_leader_optime('1')

    @patch.object(consul.Consul.Session, 'renew', Mock())
@@ -9,11 +9,11 @@ from patroni.ctl import ctl, store_config, load_config, output_members, get_dcs,
    get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException, apply_config_changes, \
    format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover, Failover
+from patroni.psycopg import OperationalError
 from patroni.utils import tzutc
-from psycopg2 import OperationalError
 from urllib3 import PoolManager

-from . import MockConnect, MockCursor, MockResponse, psycopg2_connect
+from . import MockConnect, MockCursor, MockResponse, psycopg_connect
 from .test_etcd import etcd_read, socket_getaddrinfo
 from .test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, \
    get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader, get_cluster, Member
@@ -48,7 +48,7 @@ class TestCtl(unittest.TestCase):
            self.assertRaises(PatroniCtlException, load_config, './non-existing-config-file', None)
            self.assertRaises(PatroniCtlException, load_config, './non-existing-config-file', None)

-    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.psycopg.connect', psycopg_connect)
    def test_get_cursor(self):
        self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), {}, role='master'))

@@ -57,7 +57,7 @@ class TestCtl(unittest.TestCase):
        # MockCursor returns pg_is_in_recovery as false
        self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), {}, role='replica'))

-        self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), {'database': 'foo'}, role='any'))
+        self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), {'dbname': 'foo'}, role='any'))

    def test_parse_dcs(self):
        assert parse_dcs(None) is None
@@ -165,7 +165,7 @@ class TestCtl(unittest.TestCase):
    def test_get_dcs(self):
        self.assertRaises(PatroniCtlException, get_dcs, {'dummy': {}}, 'dummy')

-    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.psycopg.connect', psycopg_connect)
    @patch('patroni.ctl.query_member', Mock(return_value=([['mock column']], None)))
    @patch('patroni.ctl.get_dcs')
    @patch.object(etcd.Client, 'read', etcd_read)
@@ -4,7 +4,7 @@ import socket
 import unittest

 from dns.exception import DNSException
-from mock import Mock, patch
+from mock import Mock, PropertyMock, patch
 from patroni.dcs.etcd import AbstractDCS, EtcdClient, Cluster, Etcd, EtcdError, DnsCachingResolver
 from patroni.exceptions import DCSError
 from patroni.utils import Retry
@@ -277,7 +277,9 @@ class TestEtcd(unittest.TestCase):
        self.etcd._base_path = '/service/failed'
        self.assertFalse(self.etcd.attempt_to_acquire_leader())

+    @patch.object(Cluster, 'min_version', PropertyMock(return_value=(2, 0)))
    def test_write_leader_optime(self):
+        self.etcd.get_cluster()
        self.etcd.write_leader_optime('0')

    def test_update_leader(self):
@@ -321,3 +323,6 @@ class TestEtcd(unittest.TestCase):

    def test_set_history_value(self):
        self.assertFalse(self.etcd.set_history_value('{}'))
+
+    def test_last_seen(self):
+        self.assertIsNotNone(self.etcd.last_seen)
@@ -19,7 +19,7 @@ from patroni.utils import tzutc
 from patroni.watchdog import Watchdog
 from six.moves import builtins

-from . import PostgresInit, MockPostmaster, psycopg2_connect, requests_get
+from . import PostgresInit, MockPostmaster, psycopg_connect, requests_get
 from .test_etcd import socket_getaddrinfo, etcd_read, etcd_write

 SYSID = '12345678901'
@@ -80,13 +80,14 @@ def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None):
    )


-def get_node_status(reachable=True, in_recovery=True, timeline=2,
-                    wal_position=10, nofailover=False, watchdog_failed=False):
+def get_node_status(reachable=True, in_recovery=True, dcs_last_seen=0,
+                    timeline=2, wal_position=10, nofailover=False,
+                    watchdog_failed=False):
    def fetch_node_status(e):
        tags = {}
        if nofailover:
            tags['nofailover'] = True
-        return _MemberStatus(e, reachable, in_recovery, timeline, wal_position, tags, watchdog_failed)
+        return _MemberStatus(e, reachable, in_recovery, dcs_last_seen, timeline, wal_position, tags, watchdog_failed)
    return fetch_node_status


@@ -322,7 +323,7 @@ class TestHa(PostgresInit):
        self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader because I had the session lock')

-    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.psycopg.connect', psycopg_connect)
    def test_acquire_lock_as_master(self):
        self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')

@@ -349,7 +350,7 @@ class TestHa(PostgresInit):
        self.ha.has_lock = true
        self.p.is_leader = false
        self.p.set_role('master')
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')

    def test_demote_after_failing_to_obtain_lock(self):
        self.ha.acquire_lock = false
@@ -389,7 +390,7 @@ class TestHa(PostgresInit):
        self.ha.cluster = get_cluster_initialized_with_leader()
        self.ha.cluster.is_unlocked = false
        self.ha.has_lock = true
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')

    def test_demote_because_not_having_lock(self):
        self.ha.cluster.is_unlocked = false
@@ -401,6 +402,8 @@ class TestHa(PostgresInit):
        self.ha.has_lock = true
        self.ha.update_lock = false
        self.assertEqual(self.ha.run_cycle(), 'demoted self because failed to update leader lock in DCS')
+        with patch.object(Ha, '_get_node_to_follow', Mock(side_effect=DCSError('foo'))):
+            self.assertEqual(self.ha.run_cycle(), 'demoted self because failed to update leader lock in DCS')
        self.p.is_leader = false
        self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS')

@@ -408,16 +411,16 @@ class TestHa(PostgresInit):
    def test_follow(self):
        self.ha.cluster.is_unlocked = false
        self.p.is_leader = false
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()')
        self.ha.patroni.replicatefrom = "foo"
        self.p.config.check_recovery_conf = Mock(return_value=(True, False))
        self.ha.cluster.config.data.update({'slots': {'l': {'database': 'a', 'plugin': 'b'}}})
        self.ha.cluster.members[1].data['tags']['replicatefrom'] = 'postgresql0'
        self.ha.patroni.nofailover = True
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()')
        del self.ha.cluster.config.data['slots']
        self.ha.cluster.config.data.update({'postgresql': {'use_slots': False}})
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()')
        del self.ha.cluster.config.data['postgresql']['use_slots']

    def test_follow_in_pause(self):
@@ -457,6 +460,8 @@ class TestHa(PostgresInit):
        self.ha.cluster = get_cluster_not_initialized_without_leader()
        self.assertEqual(self.ha.bootstrap(), 'failed to acquire initialize lock')

+    @patch('patroni.psycopg.connect', psycopg_connect)
+    @patch.object(Postgresql, 'connection', Mock(return_value=None))
    def test_bootstrap_initialized_new_cluster(self):
        self.ha.cluster = get_cluster_not_initialized_without_leader()
        self.e.initialize = true
@@ -474,6 +479,8 @@ class TestHa(PostgresInit):
        self.p.is_running = false
        self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)

+    @patch('patroni.psycopg.connect', psycopg_connect)
+    @patch.object(Postgresql, 'connection', Mock(return_value=None))
    def test_bootstrap_release_initialize_key_on_watchdog_failure(self):
        self.ha.cluster = get_cluster_not_initialized_without_leader()
        self.e.initialize = true
@@ -484,7 +491,7 @@ class TestHa(PostgresInit):
            self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap')
            self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)

-    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.psycopg.connect', psycopg_connect)
    def test_reinitialize(self):
        self.assertIsNotNone(self.ha.reinitialize())

@@ -536,27 +543,27 @@ class TestHa(PostgresInit):
        self.ha.fetch_node_status = get_node_status()
        self.ha.has_lock = true
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
        f = Failover(0, self.p.name, '', None)
        self.ha.cluster = get_cluster_initialized_with_leader(f)
        self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
        self.ha._rewind.rewind_or_reinitialize_needed_and_possible = true
        self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
        self.ha.fetch_node_status = get_node_status(nofailover=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
        self.ha.fetch_node_status = get_node_status(watchdog_failed=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
        self.ha.fetch_node_status = get_node_status(timeline=1)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
        self.ha.fetch_node_status = get_node_status(wal_position=1)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
        # manual failover from the previous leader to us won't happen if we hold the nofailover flag
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')

        # Failover scheduled time must include timezone
        scheduled = datetime.datetime.now()
@@ -565,28 +572,28 @@ class TestHa(PostgresInit):

        scheduled = datetime.datetime.utcnow().replace(tzinfo=tzutc)
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())

        scheduled = scheduled + datetime.timedelta(seconds=30)
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())

        scheduled = scheduled + datetime.timedelta(seconds=-600)
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())

        scheduled = None
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())

    def test_manual_failover_from_leader_in_pause(self):
        self.ha.has_lock = true
        self.ha.is_paused = true
        scheduled = datetime.datetime.now()
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('PAUSE: no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
-        self.assertEqual('PAUSE: no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())

    def test_manual_failover_from_leader_in_synchronous_mode(self):
        self.p.is_leader = true
@@ -595,7 +602,7 @@ class TestHa(PostgresInit):
        self.ha.is_failover_possible = false
        self.ha.process_sync_replication = Mock()
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, None))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, 'a'))
        self.ha.is_failover_possible = true
        self.assertEqual('manual failover: demoting myself', self.ha.run_cycle())
@@ -623,6 +630,11 @@ class TestHa(PostgresInit):
        # same as previous, but set the current member to nofailover. In no case it should be elected as a leader
        self.ha.patroni.nofailover = True
        self.assertEqual(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote')
+        # in sync mode only the sync node is allowed to take over
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'other', None))
+        self.ha.patroni.nofailover = False
+        self.ha.is_synchronous_mode = true
+        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')

    def test_manual_failover_process_no_leader_in_pause(self):
        self.ha.is_paused = true
@@ -756,7 +768,7 @@ class TestHa(PostgresInit):
        self.p.config.check_recovery_conf = Mock(return_value=(False, False))
        self.ha._leader_timeline = 1
        self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (leader) the standby leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (leader), the standby leader with the lock')
        self.p.set_role('replica')
        self.p.config.check_recovery_conf = Mock(return_value=(True, False))
        self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
@@ -766,7 +778,7 @@ class TestHa(PostgresInit):
        self.p.name = 'replica'
        self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
        self.assertEqual(self.ha.run_cycle(),
-                         'no action. I am a secondary (replica) and following a standby leader (leader)')
+                         'no action. I am (replica), a secondary, and following a standby leader (leader)')
        with patch.object(Leader, 'conn_url', PropertyMock(return_value='')):
            self.assertEqual(self.ha.run_cycle(), 'continue following the old known standby leader')

@@ -860,7 +872,7 @@ class TestHa(PostgresInit):
        self.ha.has_lock = false
        self.p.is_leader = false
        self.assertEqual(self.ha.run_cycle(),
-                         'no action. I am a secondary (postgresql0) and following a leader (leader)')
+                         'no action. I am (postgresql0), a secondary, and following a leader (leader)')
        check_calls([(update_lock, False), (demote, False)])

    def test_manual_failover_while_starting(self):
@@ -1085,7 +1097,7 @@ class TestHa(PostgresInit):
        self.ha.cluster.config.data.clear()
        self.ha.has_lock = true
        self.ha.cluster.is_unlocked = false
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')

    def test_watch(self):
        self.ha.cluster = get_cluster_initialized_with_leader()
@@ -1097,6 +1109,14 @@ class TestHa(PostgresInit):
    def test_shutdown(self):
        self.p.is_running = false
        self.ha.is_leader = true
+
+        def stop(*args, **kwargs):
+            kwargs['on_shutdown'](123)
+
+        self.p.stop = stop
+        self.ha.shutdown()
+
+        self.ha.is_failover_possible = true
        self.ha.shutdown()

    @patch('time.sleep', Mock())
@@ -1120,7 +1140,7 @@ class TestHa(PostgresInit):
        self.ha.cluster.is_unlocked = false
        for tl in (1, 3):
            self.p.get_master_timeline = Mock(return_value=tl)
-            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')

    @patch('sys.exit', return_value=1)
    def test_abort_join(self, exit_mock):
@@ -1133,11 +1153,11 @@ class TestHa(PostgresInit):
        self.ha.has_lock = true
        self.ha.cluster.is_unlocked = false
        self.ha.is_paused = true
-        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0), the leader with the lock')
        self.ha.is_paused = false
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')

-    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.psycopg.connect', psycopg_connect)
    def test_permanent_logical_slots_after_promote(self):
        config = ClusterConfig(1, {'slots': {'l': {'database': 'postgres', 'plugin': 'test_decoding'}}}, 1)
        self.p.name = 'other'
@@ -1145,7 +1165,7 @@ class TestHa(PostgresInit):
        self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
        self.ha.cluster = get_cluster_initialized_without_leader(leader=True, cluster_config=config)
        self.ha.has_lock = true
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (other) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (other), the leader with the lock')

    @patch.object(Cluster, 'has_member', true)
    def test_run_cycle(self):
@@ -1169,7 +1189,7 @@ class TestHa(PostgresInit):
        self.ha.has_lock = true
        self.assertEqual(self.ha.run_cycle(), 'PAUSE: released leader key voluntarily due to the system ID mismatch')

-    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.psycopg.connect', psycopg_connect)
    @patch('os.path.exists', Mock(return_value=True))
    @patch('shutil.rmtree', Mock())
    @patch('os.makedirs', Mock())
@@ -1184,3 +1204,8 @@ class TestHa(PostgresInit):
        self.ha.cluster.is_unlocked = false
        self.p.is_leader = false
        self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots'))
+
+    def test_is_failover_possible(self):
+        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))
@@ -13,15 +13,23 @@ from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
-from patroni import Patroni, main as _main, patroni_main, check_psycopg2
+from patroni import Patroni, main as _main, patroni_main, check_psycopg
 from six.moves import BaseHTTPServer, builtins
 from threading import Thread

-from . import psycopg2_connect, SleepException
+from . import psycopg_connect, SleepException
 from .test_etcd import etcd_read, etcd_write
 from .test_postgresql import MockPostmaster


+def mock_import(*args, **kwargs):
+    if args[0] == 'psycopg':
+        raise ImportError
+    ret = Mock()
+    ret.__version__ = '2.5.3.dev1 a b c'
+    return ret
+
+
 class MockFrozenImporter(object):

    toc = set(['patroni.dcs.etcd'])
@@ -29,7 +37,7 @@ class MockFrozenImporter(object):

@patch('time.sleep', Mock())
@patch('subprocess.call', Mock(return_value=0))
-@patch('psycopg2.connect', psycopg2_connect)
+@patch('patroni.psycopg.connect', psycopg_connect)
@patch.object(ConfigHandler, 'append_pg_hba', Mock())
@patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
@patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -181,8 +189,8 @@ class TestPatroni(unittest.TestCase):
        self.p.ha.shutdown = Mock(side_effect=Exception)
        self.p.shutdown()

-    def test_check_psycopg2(self):
+    def test_check_psycopg(self):
        with patch.object(builtins, '__import__', Mock(side_effect=ImportError)):
-            self.assertRaises(SystemExit, check_psycopg2)
-        with patch('psycopg2.__version__', '2.5.3.dev1 a b c'):
-            self.assertRaises(SystemExit, check_psycopg2)
+            self.assertRaises(SystemExit, check_psycopg)
+        with patch.object(builtins, '__import__', mock_import):
+            self.assertRaises(SystemExit, check_psycopg)
@@ -1,12 +1,14 @@
 import datetime
 import os
 import psutil
-import psycopg2
 import re
 import subprocess
 import time

 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
+
+import patroni.psycopg as psycopg
+
 from patroni.async_executor import CriticalTask
 from patroni.dcs import Cluster, RemoteMember, SyncState
 from patroni.exceptions import PostgresConnectionException, PatroniException
@@ -17,7 +19,7 @@ from patroni.utils import RetryFailedError
 from six.moves import builtins
 from threading import Thread, current_thread

-from . import BaseTestPostgresql, MockCursor, MockPostmaster, psycopg2_connect
+from . import BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect


 mtime_ret = {}
@@ -87,7 +89,7 @@ Data page checksum version:           0


@patch('subprocess.call', Mock(return_value=0))
-@patch('psycopg2.connect', psycopg2_connect)
+@patch('patroni.psycopg.connect', psycopg_connect)
 class TestPostgresql(BaseTestPostgresql):

    @patch('subprocess.call', Mock(return_value=0))
@@ -203,6 +205,21 @@ class TestPostgresql(BaseTestPostgresql):
        mock_postmaster.signal_stop.side_effect = [None, True]
        self.assertTrue(self.p.stop(on_safepoint=mock_callback, stop_timeout=30))

+    @patch('time.sleep', Mock())
+    @patch.object(Postgresql, 'is_running', MockPostmaster)
+    @patch.object(Postgresql, '_wait_for_connection_close', Mock())
+    @patch.object(Postgresql, 'latest_checkpoint_location', Mock(return_value='7'))
+    def test__do_stop(self):
+        mock_callback = Mock()
+        with patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down'})):
+            self.assertTrue(self.p.stop(on_shutdown=mock_callback, stop_timeout=3))
+            mock_callback.assert_called()
+        with patch.object(Postgresql, 'controldata',
+                          Mock(return_value={'Database cluster state': 'shut down in recovery'})):
+            self.assertTrue(self.p.stop(on_shutdown=mock_callback, stop_timeout=3))
+        with patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shutting down'})):
+            self.assertTrue(self.p.stop(on_shutdown=mock_callback, stop_timeout=3))
+
    def test_restart(self):
        self.p.start = Mock(return_value=False)
        self.assertFalse(self.p.restart())
@@ -304,7 +321,7 @@ class TestPostgresql(BaseTestPostgresql):
        m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}})
        self.p.follow(m)

-    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError))
+    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError))
    def test__query(self):
        self.assertRaises(PostgresConnectionException, self.p._query, 'blabla')
        self.p._state = 'restarting'
@@ -313,14 +330,14 @@ class TestPostgresql(BaseTestPostgresql):
    def test_query(self):
        self.p.query('select 1')
        self.assertRaises(PostgresConnectionException, self.p.query, 'RetryFailedError')
-        self.assertRaises(psycopg2.ProgrammingError, self.p.query, 'blabla')
+        self.assertRaises(psycopg.ProgrammingError, self.p.query, 'blabla')

    @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT))
    def test_is_leader(self):
        self.assertTrue(self.p.is_leader())
        self.p.reset_cluster_info_state(None)
        with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))):
-            self.assertRaises(PostgresConnectionException, self.p.is_leader)
+            self.assertFalse(self.p.is_leader())

    @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down',
                                                                'Latest checkpoint location': '0/1ADBC18',
@@ -415,7 +432,7 @@ class TestPostgresql(BaseTestPostgresql):
    @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
    def test_is_leader_exception(self):
        self.p.start()
-        self.p.query = Mock(side_effect=psycopg2.OperationalError("not supported"))
+        self.p.query = Mock(side_effect=psycopg.OperationalError("not supported"))
        self.assertTrue(self.p.stop())

    @patch('os.rename', Mock())
@@ -544,7 +561,7 @@ class TestPostgresql(BaseTestPostgresql):
            t.start()
            t.join()

-        with patch.object(MockCursor, "execute", side_effect=psycopg2.Error):
+        with patch.object(MockCursor, "execute", side_effect=psycopg.Error):
            self.assertIsNone(self.p.postmaster_start_time())

    def test_check_for_startup(self):
@@ -707,7 +724,7 @@ class TestPostgresql(BaseTestPostgresql):
            self.p.stop(on_safepoint=mock_callback)

            mock_postmaster.is_running.side_effect = [True, False, False]
-            with patch.object(MockCursor, "execute", Mock(side_effect=psycopg2.Error)):
+            with patch.object(MockCursor, "execute", Mock(side_effect=psycopg.Error)):
                self.p.stop(on_safepoint=mock_callback)

    def test_terminate_starting_postmaster(self):
@@ -5,7 +5,7 @@ from patroni.postgresql.cancellable import CancellableSubprocess
 from patroni.postgresql.rewind import Rewind
 from six.moves import builtins

-from . import BaseTestPostgresql, MockCursor, psycopg2_connect
+from . import BaseTestPostgresql, MockCursor, psycopg_connect


 class MockThread(object):
@@ -47,7 +47,7 @@ def mock_single_user_mode(self, communicate, options):


@patch('subprocess.call', Mock(return_value=0))
-@patch('psycopg2.connect', psycopg2_connect)
+@patch('patroni.psycopg.connect', psycopg_connect)
 class TestRewind(BaseTestPostgresql):

    def setUp(self):
@@ -141,13 +141,15 @@ class TestRewind(BaseTestPostgresql):
        self.leader = self.leader.member
        self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
        mock_check_leader_is_not_in_recovery.return_value = True
-        self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+        self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+        self.r.reset_state()
        self.r.trigger_check_diverged_lsn()
-        with patch('psycopg2.connect', Mock(side_effect=Exception)):
+        with patch('patroni.psycopg.connect', Mock(side_effect=Exception)):
            self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
        self.r.trigger_check_diverged_lsn()
-        with patch.object(MockCursor, 'fetchone', Mock(side_effect=[('', 3, '0/0'), ('', b'3\t0/40159C0\tn\n')])):
-            self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+        with patch.object(MockCursor, 'fetchone', Mock(side_effect=[('', 3, '0/0'), ('', b'1\t0/40159C0\tn\n')])):
+            self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+        self.r.reset_state()
        self.r.trigger_check_diverged_lsn()
        with patch.object(MockCursor, 'fetchone', Mock(return_value=('', 1, '0/0'))):
            with patch.object(Rewind, '_get_local_timeline_lsn', Mock(return_value=(True, 1, '0/0'))):
@@ -219,18 +221,20 @@ class TestRewind(BaseTestPostgresql):
    @patch('patroni.postgresql.rewind.Thread', MockThread)
    @patch.object(Postgresql, 'controldata')
    @patch.object(Postgresql, 'checkpoint')
-    def test_ensure_checkpoint_after_promote(self, mock_checkpoint, mock_controldata):
-        mock_checkpoint.return_value = None
+    @patch.object(Postgresql, 'get_master_timeline')
+    def test_ensure_checkpoint_after_promote(self, mock_get_master_timeline, mock_checkpoint, mock_controldata):
+        mock_controldata.return_value = {"Latest checkpoint's TimeLineID": 1}
+        mock_get_master_timeline.return_value = 1
+        self.r.ensure_checkpoint_after_promote(Mock())
+
+        self.r.reset_state()
+        mock_get_master_timeline.return_value = 2
+        mock_checkpoint.return_value = 0
        self.r.ensure_checkpoint_after_promote(Mock())
        self.r.ensure_checkpoint_after_promote(Mock())

        self.r.reset_state()
-        mock_controldata.return_value = {"Latest checkpoint's TimeLineID": 1}
+
+        mock_controldata.side_effect = TypeError
        mock_checkpoint.side_effect = Exception
        self.r.ensure_checkpoint_after_promote(Mock())
-        self.r.ensure_checkpoint_after_promote(Mock())
-
-        self.r.reset_state()
-        mock_controldata.side_effect = TypeError
-        self.r.ensure_checkpoint_after_promote(Mock())
-        self.r.ensure_checkpoint_after_promote(Mock())
@@ -1,20 +1,20 @@
 import mock
 import os
-import psycopg2
 import unittest


 from mock import Mock, PropertyMock, patch

+from patroni import psycopg
 from patroni.dcs import Cluster, ClusterConfig, Member
 from patroni.postgresql import Postgresql
 from patroni.postgresql.slots import SlotsHandler, fsync_dir

-from . import BaseTestPostgresql, psycopg2_connect, MockCursor
+from . import BaseTestPostgresql, psycopg_connect, MockCursor


@patch('subprocess.call', Mock(return_value=0))
-@patch('psycopg2.connect', psycopg2_connect)
+@patch('patroni.psycopg.connect', psycopg_connect)
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
 class TestSlotsHandler(BaseTestPostgresql):

@@ -34,7 +34,7 @@ class TestSlotsHandler(BaseTestPostgresql):
                                   'ignore_slots': [{'name': 'blabla'}]}, 1)
        cluster = Cluster(True, config, self.leader, 0,
                          [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10})
-        with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)):
+        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')
        self.s.sync_replication_slots(cluster, False)
@@ -86,14 +86,14 @@ class TestSlotsHandler(BaseTestPostgresql):
                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
        self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.errors.UndefinedFile)):
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.UndefinedFile)):
            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
        cluster.slots['ls'] = 'a'
        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])

-    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError))
+    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError))
    def test_copy_logical_slots(self):
        self.s.copy_logical_slots(self.leader, ['foo'])

@@ -1,14 +1,15 @@
-import psycopg2
 import subprocess
 import unittest

+import patroni.psycopg as psycopg
+
 from mock import Mock, PropertyMock, patch, mock_open
 from patroni.scripts import wale_restore
 from patroni.scripts.wale_restore import WALERestore, main as _main, get_major_version
 from six.moves import builtins
 from threading import current_thread

-from . import MockConnect, psycopg2_connect
+from . import MockConnect, psycopg_connect

 wale_output_header = (
    b'name\tlast_modified\t'
@@ -34,7 +35,7 @@ WALE_TEST_RETRIES = 2
@patch('os.makedirs', Mock(return_value=True))
@patch('os.path.exists', Mock(return_value=True))
@patch('os.path.isdir', Mock(return_value=True))
-@patch('psycopg2.connect', psycopg2_connect)
+@patch('patroni.psycopg.connect', psycopg_connect)
@patch('subprocess.check_output', Mock(return_value=wale_output))
 class TestWALERestore(unittest.TestCase):

@@ -57,7 +58,7 @@ class TestWALERestore(unittest.TestCase):
        with patch('subprocess.check_output', Mock(return_value=wale_output.replace(b'167772160', b'1'))):
            self.assertFalse(self.wale_restore.should_use_s3_to_create_replica())

-        with patch('psycopg2.connect', Mock(side_effect=psycopg2.Error("foo"))):
+        with patch('patroni.psycopg.connect', Mock(side_effect=psycopg.Error("foo"))):
            save_no_master = self.wale_restore.no_master
            save_master_connection = self.wale_restore.master_connection

@@ -6,8 +6,8 @@ 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 mock import Mock, patch
-from patroni.dcs.zookeeper import Leader, PatroniKazooClient,\
+from mock import Mock, PropertyMock, patch
+from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\
        PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError


@@ -144,7 +144,8 @@ class TestZooKeeper(unittest.TestCase):
    @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient)
    def setUp(self):
        self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test',
-                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10})
+                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10,
+                             'set_acls': {'CN=principal2': ['ALL']}})

    def test_session_listener(self):
        self.zk.session_listener(KazooState.SUSPENDED)
@@ -173,6 +174,8 @@ class TestZooKeeper(unittest.TestCase):
        self.assertRaises(ZooKeeperError, self.zk.get_cluster)
        cluster = self.zk.get_cluster(True)
        self.assertIsInstance(cluster.leader, Leader)
+        self.zk.status_watcher(None)
+        self.zk.get_cluster()
        self.zk.touch_member({'foo': 'foo'})
        self.zk._name = 'bar'
        self.zk.status_watcher(None)
@@ -213,6 +216,7 @@ class TestZooKeeper(unittest.TestCase):
        self.zk.touch_member({'retry': 'retry'})
        self.zk._fetch_cluster = True
        self.zk.get_cluster()
+        self.zk.touch_member({'retry': 'retry'})
        self.zk.touch_member({'conn_url': 'postgres://repuser:rep-pass@localhost:5434/postgres',
                              'api_url': 'http://127.0.0.1:8009/patroni'})

@@ -224,6 +228,7 @@ class TestZooKeeper(unittest.TestCase):
    def test_update_leader(self):
        self.assertTrue(self.zk.update_leader(12345))

+    @patch.object(Cluster, 'min_version', PropertyMock(return_value=(2, 0)))
    def test_write_leader_optime(self):
        self.zk.last_lsn = '0'
        self.zk.write_leader_optime('1')
@@ -232,6 +237,7 @@ class TestZooKeeper(unittest.TestCase):
        with patch.object(MockKazooClient, 'set_async', Mock()):
            self.zk.write_leader_optime('2')
        self.zk._base_path = self.zk._base_path.replace('test', 'bla')
+        self.zk.get_cluster()
        self.zk.write_leader_optime('3')

    def test_delete_cluster(self):
Author	SHA1	Message	Date
Alexander KukushkinandGitHub	dc9ff4cb8a	Release 2.1.2 (#2136 ) * Implement missing unit-tests * Bump version * Update release notes	2021-12-03 15:49:57 +01:00
Alexander KukushkinandGitHub	d7dc3c2d96	Handle missing timelines in history file when deciding to rewind (#2120 ) When restore_command is configured Postgres is trying to fetch/apply all possible WAL segments and also fetch history files in order to select the correct timeline. It could result in a situation where the new history file will be missing some timelines. Example: - node1 demotes/crashes on timeline 1 - node2 promotes to timeline 2 and archives `00000002.history` and crashes - node1 recovers as a replica, "replays" `00000002.history` and promotes to timeline 3 As a result, the `00000003.history` will not have the line with timeline 2, because it never replayed any WAL segment from it. The `pg_rewind` tool is supposed to correctly handle such case when rewinding node2 from node1, but Patroni when deciding whether the rewind should happen was searching for the exact timeline in the history file from the new primary. The solution is to assume that rewind is required if the current replica timeline is missing. In addition to that this PR makes sure that the primary isn't running in recovery before starting the procedure of rewind check. Close https://github.com/zalando/patroni/issues/2118 and https://github.com/zalando/patroni/issues/2124	2021-12-02 11:35:30 +01:00
Michael BanckandGitHub	90b3736fec	Remove duplicate hosts from the etcd machine cache (#2127 ) Close #2126	2021-12-02 11:35:11 +01:00
Alexander KukushkinandGitHub	63ee42a85c	Clear event on the leader node when /status was updated (#2125 ) Not doing so causing excessive HA loop runs with Zookeeper. This moment wasn't fixed correctly in the #1875	2021-11-30 16:33:38 +01:00
Alexander KukushkinandGitHub	d24051c31c	Optimize case when we don't have permanent logical slots (#2121 ) The unnecessary call of SlotsHandler.process_permanent_slots() results in one additional query to `pg_replication_slots` view every HA loop.	2021-11-30 14:20:55 +01:00
Alexander KukushkinandGitHub	31d7540cc5	Prefer members without nofailover when picking sync nodes (#2108 ) Previously sync nodes were selected only based on replication lag and hence the node with `nofailover` tag had the same chances to become synchronous as any other node. That behavior was confusing and dangerous at the same time, because in case of failed primary the failover couldn't happen automatically. Close https://github.com/zalando/patroni/issues/2089	2021-11-30 14:20:03 +01:00
Alexander KukushkinandGitHub	256a359a1e	Fix a litle bug around psycopg 3.0 (#2123 ) Cursor.execute now returns cursor itself, while in psycopg2 it was returning None	2021-11-19 16:29:39 +01:00
Alexander KukushkinandGitHub	17e523b175	Optimize checkpoint after promote (#2114 ) 1. Avoid doing CHECKPOINT if `pg_control` is already updated. 2. Explicitly call ensure_checkpoint_after_promote() right after the bootstrap finished successfully.	2021-11-19 14:33:24 +01:00
Alexander KukushkinandGitHub	fce889cd04	Compatibility with psycopg 3.0 (#2088 ) By default `psycopg2` is preferred. The `psycopg>=3.0` will be used only if `psycopg2` is not available or its version is too old.	2021-11-19 14:32:54 +01:00
Alexander KukushkinandGitHub	edfe2a84e9	Fix a few issues with Patroni API (#2116 ) 1. The `client_address` tuple may have more than two elements in case of IPv6 2. Return `cluster_unlocked` only when the value is true and handle it respectively in the do_GET_metrics() 3. Return `cluster_unlocked` and `dcs_last_seen` even if Postgres isn't running/queries timing out Close https://github.com/zalando/patroni/issues/2113	2021-11-12 15:02:53 +01:00
Alexander KukushkinandGitHub	00d125c512	Avoid unnecessary updates of the members ZNode. (#2115 ) When deciding whether the ZNode should be updated we rely on the cached version of the cluster, which is updated only when members ZNodes are deleted/created or the `/status`, `/sync`, `/failover`, `/config`, or `/history` ZNodes are updated. I.e. after the update of the current member ZNode succeeded the cache becomes stale and all further updates are always performed even if the value didn't change. In order to solve it, we introduce the new attribute in the Zookeeper class and will use it for memorizing the actual value and for later comparison.	2021-11-12 15:00:54 +01:00
Alexander KukushkinandGitHub	fd1e0f1c1b	BUGFIX: use_unix_socket_repl didn't work is some cases (#2103 ) Specifically, if `postgresql.unix_socket_directories` is not set. In this case Patroni is supposed to use only the port in the connection string, but the `get_replication_connection_cursor()` method defaulted to host='localhost'	2021-10-29 12:09:38 +02:00
Alwyn DavisandGitHub	14bb28c349	Allow setting ACLs for znodes in Zookeeper (#2086 ) Add a configuration option (`set_acls`) for Zookeeper DCS so that Kazoo will apply a default ACL for each znode that it creates. The intention is to improve security of the znodes when a single Zookeeper cluster is used as the DCS for multiple Patroni clusters. Zookeeper [does not apply an ACL to child znodes](https://zookeeper.apache.org/doc/current/zookeeperProgrammers.html#sc_ZooKeeperAccessControl), so permissions can't be set at the `scope` level and then be inherited by other znodes that Patroni creates. Kazoo instead [provides an option for configuring a default_acl](https://kazoo.readthedocs.io/en/latest/api/client.html#kazoo.client.KazooClient.__init__) that will be applied on node creation. Example configuration in Patroni might then be: ``` zookeeper: set_acls: CN=principal1: [ALL] CN=principal2: - READ ```	2021-10-28 09:59:45 +02:00
Michael BanckandGitHub	95479526ef	Fix typo (#2091 )	2021-10-26 15:12:10 +02:00
Alexander KukushkinandGitHub	47ebda0d5d	Fix a few issues in kubernetes.py (#2084 ) 1. Two `TypeError`-s raised from `ApiClient.request()` method 2. Use the _retry() wrapper function instead of callable object in the `_update_leader_with_retry()` when trying to workaround concurrent updates of the leader object.	2021-10-08 16:13:28 +02:00
Nicolas PAYARTandGitHub	64ae2bb885	Add compatibility with PG 14 in README (#2083 ) It was just missing there	2021-10-08 15:51:55 +02:00
Alexander KukushkinandGitHub	250328b84b	Use cached role as a fallback when postgres is slow (#2082 ) In some extreme cases Postgres could be so slow that the normal monitoring query doesn't finish in a few seconds. It results in the exception being raised from the `Postgresql._cluster_info_state_get()` method, which could lead to the situation that postgres isn't demoted on time. In order to make it reliable we will catch the exception and use the cached state of postgres (`is_running()` and `role`) to determine whether postgres is running as a primary. Close https://github.com/zalando/patroni/issues/2073	2021-10-07 16:08:21 +02:00
Alexander KukushkinandGitHub	89388c2e4b	Handle DCS exceptions when demoting (#2081 ) While doing demote due to failure to update leader lock it could happen that DCS goes completely down and the get_cluster() call raise the exception. Not being properly handled it results in postgres remaining stopped until DCS recovers.	2021-10-07 16:08:10 +02:00
Michael BanckandGitHub	e28557d2f0	Fix sphinx build. (#2080 ) Sphinx' add_stylesheet() has been deprecated for a long time and got removed in recent versions of sphinx. If available, use add_css_file() instead. Close #2079.	2021-10-07 16:07:41 +02:00
Farid ZarazvandandGitHub	34db0bba16	PostgreSQL v14 is supported since v2.1.0 (#2078 )	2021-10-07 16:07:00 +02:00
Kostiantyn NemchenkoandGitHub	3616906434	Add sslcrldir connection parameter support (#2068 ) This allows setting the `sslcrldir` connection parameter available since PostgreSQL 14.	2021-10-07 16:04:27 +02:00
Alexander KukushkinandGitHub	d394b63c9f	Release the leader lock when pg_controldata reports "shut down" (#2067 ) Due to different reasons, it could happen that WAL archiving on the primary stuck or significantly delayed. If we try to do a switchover or shut it down, the shutdown will take forever and will not finish until the whole backlog of WALs is processed. In the meantime, Patroni keeps updating the leader lock, which prevents other nodes from starting the leader race even if it is known that they received/applied all changes. The `Database cluster state:` is changed to `"shut down"` after: - all data is fsynced to disk and the latest checkpoint is written to WAL - all streaming replicas confirmed that they received all changes (including the latest checkpoint) - at the same time, the archiver process continues to do its job and the postmaster process is still running. In order to solve this problem and make the switchover more reliable/fast in a case when `archive_command` is slow/failing, Patroni will remove the leader key immediately after `pg_controldata` started reporting PGDATA as `"shut down"` cleanly and it verified that there is at least one replica that received all changes. If there are no replicas that fulfill the condition the leader key isn't removed and the old behavior is retained, i.e. Patroni will keep updating it.	2021-10-05 10:55:35 +02:00
Alexander KukushkinandGitHub	1c2bf258d6	Allow switchover only to sync nodes when synchronous replication is on (#2076 ) Close https://github.com/zalando/patroni/issues/2074	2021-10-04 16:23:45 +02:00
Alexander KukushkinandGitHub	a431f50378	Check only sync nodes when assessing failover capabilities (#2065 ) When synchronous_mode is enabled we should check only synchronous nodes in is_failover_possible().	2021-09-24 08:22:54 +02:00
Alexander KukushkinandGitHub	fca724186e	DCS.write_leader_optime() should update /status key (#2064 ) This moment was forgotten in the failover logical slots implementation.	2021-09-24 08:22:20 +02:00
Michael BanckandGitHub	2f31e88bdc	Add dcs_last_seen field to API (#2051 ) This field notes the last time (as unix epoch) a cluster member has successfully communicated with the DCS. This is useful to identify and/or analyze network partitions. Also, expose dcs_last_seen in the MemberStatus class and its from_api_response() method.	2021-09-22 10:01:35 +02:00
Jorge SolórzanoandGitHub	80c1127b70	Cast to int wal_keep_segments conversion to wal_keep_size (#2063 ) Fixes #2062	2021-09-22 10:00:42 +02:00
Alexander KukushkinandGitHub	258e7e24f4	Ensure pg_replication_slot_advance() doesn't timeout (#2060 ) The bigger gap between the slot flush LSN and the LSN we want to advance to becomes more time it takes for the call to finish. Once started failing the "lag" will grow more or less infinitely, that have the following negative side-effects: 1. Size of pg_wal on the replica will grow 2. Since the hot_standby_feedback is forcefully enabled, the primary will stop cleaning up dead tuples I.e., we are not only in danger of running out of disk space, but also increasing chances of transaction wraparound to happen. In order to mitigate it, we want to set the `statement_timeout` to 0 before calling `pg_replication_slot_advance()`. Since the call is happening from the main HA loop and could take more than `loop_wait`, the next heartbeat run could be delayed. There is also a possibility that the call could take longer than `ttl` and the member key/session in DCS for a given replica expires, but, the slot LSN in DCS is updated by the primary every `loop_wait` seconds. Hence, we don't expect that the slot_advance() call will take significantly longer than the `loop_wait` and therefore chances of the member key/session to expire are very low.	2021-09-17 16:39:39 +02:00
Michael BanckandGitHub	fae96b3148	Improve "I am" status messages (#2056 )	2021-09-17 14:46:07 +02:00
Michael BanckandGitHub	8a9e649aa1	Add log before demoting (which can take some time) (#2057 ) It can take some time for the demote to finish and it might not be obvious from looking at the logs what exactly is going on.	2021-09-17 14:45:32 +02:00
Alexander KukushkinandGitHub	7bd28250ca	Skip temporary replication slots while doing slot management (#2055 ) Starting from v10 `pg_basebackup` creates a temporary replication slot for WAL streaming and Patroni was trying to drop it because the slot name looks unknown. In order to fix it, we skip all temporary slots when querying `pg_stat_replication_slots` view. Another option to solve the problem would be running `pg_basebackup` with `--slot=current_node_name` option, but unfortunately at the moment when `pg_basebackup` is executed, we don't yet know the major version (the `--slot` option was added in v9.6). Ref: https://github.com/zalando/patroni/issues/2046#issuecomment-912521502	2021-09-17 14:44:54 +02:00
Alexander KukushkinandGitHub	21145d18d1	Delay the next attempt of recovery till next HA loop (#2054 ) If Postgres crashed due to out of disk space (for example) and fails to start because of that Patroni is too eagerly trying to recover it and producing too many logs	2021-09-17 13:46:46 +02:00