Merge branch 'master' of github.com:zalando/patroni into feature/citus-secondaries

Dockerfile

@@ -25,8 +25,7 @@ RUN set -ex \
             | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \
             | xargs apt-get install -y vim curl less jq locales haproxy sudo \
                             python3-etcd python3-kazoo python3-pip busybox \
-                            net-tools iputils-ping --fix-missing \
-    && pip3 install dumb-init \
+                            net-tools iputils-ping dumb-init --fix-missing \
 \
     # Cleanup all locales but en_US.UTF-8
     && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \
@@ -71,7 +70,7 @@ RUN set -ex \
     # Clean up all useless packages and some files
     && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \
                 libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \
-                exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \
+                exim4-config gnupg-agent dirmngr \
                 git make \
     && apt-get autoremove -y \
     && apt-get clean -y \

Dockerfile.citus

@@ -25,7 +25,7 @@ RUN set -ex \
             | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \
             | xargs apt-get install -y vim curl less jq locales haproxy sudo \
                             python3-etcd python3-kazoo python3-pip busybox \
-                            net-tools iputils-ping lsb-release --fix-missing \
+                            net-tools iputils-ping lsb-release dumb-init --fix-missing \
     && if [ $(dpkg --print-architecture) = 'arm64' ]; then \
         apt-get install -y postgresql-server-dev-$PG_MAJOR \
                            git gcc make autoconf \
@@ -42,7 +42,6 @@ RUN set -ex \
         && apt-get update -y \
         && apt-get -y install postgresql-$PG_MAJOR-citus-11.3; \
     fi \
-    && pip3 install dumb-init \
 \
     # Cleanup all locales but en_US.UTF-8
     && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \
@@ -88,7 +87,7 @@ RUN set -ex \
     # Clean up all useless packages and some files
     && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \
                 libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \
-                exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \
+                exim4-config gnupg-agent dirmngr \
                 postgresql-server-dev-$PG_MAJOR git gcc make autoconf \
                 libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \
                 libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \

docs/ENVIRONMENT.rst

@@ -112,7 +112,11 @@ Kubernetes
 -  **PATRONI\_KUBERNETES\_NAMESPACE**: (optional) Kubernetes namespace where the Patroni pod is running. Default value is `default`.
 -  **PATRONI\_KUBERNETES\_LABELS**: Labels in format ``{label1: value1, label2: value2}``. These labels will be used to find existing objects (Pods and either Endpoints or ConfigMaps) associated with the current cluster. Also Patroni will set them on every object (Endpoint or ConfigMap) it creates.
 -  **PATRONI\_KUBERNETES\_SCOPE\_LABEL**: (optional) name of the label containing cluster name. Default value is `cluster-name`.
--  **PATRONI\_KUBERNETES\_ROLE\_LABEL**: (optional) name of the label containing Postgres role (`master` or `replica`). Patroni will set this label on the pod it is running in. Default value is `role`.
+-  **PATRONI\_KUBERNETES\_ROLE\_LABEL**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``.
+-  **PATRONI\_KUBERNETES\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `master`. Default value is `master`.
+-  **PATRONI\_KUBERNETES\_FOLLOWER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `replica`. Default value is `replica`.
+-  **PATRONI\_KUBERNETES\_STANDBY\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``.
+-  **PATRONI\_KUBERNETES\_TMP\_ROLE\_LABEL**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary.
 -  **PATRONI\_KUBERNETES\_USE\_ENDPOINTS**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state.
 -  **PATRONI\_KUBERNETES\_POD\_IP**: (optional) IP address of the pod Patroni is running in. This value is required when `PATRONI_KUBERNETES_USE_ENDPOINTS` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted.
 -  **PATRONI\_KUBERNETES\_PORTS**: (optional) if the Service object has the name for the port, the same name must appear in the Endpoint object, otherwise service won't work. For example, if your service is defined as ``{Kind: Service, spec: {ports: [{name: postgresql, port: 5432, targetPort: 5432}]}}``, then you have to set ``PATRONI_KUBERNETES_PORTS='[{"name": "postgresql", "port": 5432}]'`` and Patroni will use it for updating subsets of the leader Endpoint. This parameter is used only if `PATRONI_KUBERNETES_USE_ENDPOINTS` is set.
@@ -196,10 +200,19 @@ REST API
 -  **PATRONI\_RESTAPI\_HTTPS\_EXTRA\_HEADERS**: (optional) HTTPS headers let the REST API server pass additional information with an HTTP response when TLS is enabled. This will also pass additional information set in ``http_extra_headers``.
 -  **PATRONI\_RESTAPI\_REQUEST\_QUEUE\_SIZE**: (optional): Sets request queue size for TCP socket used by Patroni REST API.  Once the queue is full, further requests get a "Connection denied" error. The default value is 5.
 
+.. warning::
+
+    - The ``PATRONI_RESTAPI_CONNECT_ADDRESS`` must be accessible from all nodes of a given Patroni cluster. Internally Patroni is using it during the leader race to find nodes with minimal replication lag.
+    - If you enabled client certificates validation (``PATRONI_RESTAPI_VERIFY_CLIENT`` is set to ``required``), you also **must** provide **valid client certificates** in the ``PATRONI_CTL_CERTFILE``, ``PATRONI_CTL_KEYFILE``, ``PATRONI_CTL_KEYFILE_PASSWORD``. If not provided, Patroni will not work correctly.
+
+
 CTL
 ---
--  **PATRONICTL\_CONFIG\_FILE**: location of the configuration file.
--  **PATRONI\_CTL\_INSECURE**: Allow connections to REST API without verifying SSL certs.
--  **PATRONI\_CTL\_CACERT**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter.
--  **PATRONI\_CTL\_CERTFILE**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter.
--  **PATRONI\_CTL\_KEYFILE**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter.
+-  **PATRONICTL\_CONFIG\_FILE**: (optional) location of the configuration file.
+-  **PATRONI\_CTL\_USERNAME**: (optional) Basic-auth username for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "username" parameter.
+-  **PATRONI\_CTL\_PASSWORD**: (optional) Basic-auth password for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "password" parameter.
+-  **PATRONI\_CTL\_INSECURE**: (optional) Allow connections to REST API without verifying SSL certs.
+-  **PATRONI\_CTL\_CACERT**: (optional) Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter.
+-  **PATRONI\_CTL\_CERTFILE**: (optional) Specifies the file with the client certificate in the PEM format.
+-  **PATRONI\_CTL\_KEYFILE**: (optional) Specifies the file with the client secret key in the PEM format.
+-  **PATRONI\_CTL\_KEYFILE\_PASSWORD**: (optional) Specifies a password for decrypting the client keyfile.

docs/dynamic_configuration.rst

@@ -46,9 +46,9 @@ In order to change the dynamic configuration you can use either ``patronictl edi
    -  **archive\_cleanup\_command**: cleanup command for standby leader
    -  **recovery\_min\_apply\_delay**: how long to wait before actually apply WAL records on a standby leader
 
--  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
+-  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. The physical slots are maintained only in the current primary. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
 
-   -  **my\_slot\_name**: the name of replication slot. If the permanent slot name matches with the name of the current primary it will not be created. Everything else is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots.
+   -  **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster <existing_data>` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni.
 
       -  **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``.
       -  **database**: the database name where logical slots should be created.

docs/kubernetes.rst

@@ -32,6 +32,55 @@ Configuration
 
 Patroni Kubernetes :ref:`settings <kubernetes_settings>` and :ref:`environment variables <kubernetes_environment>` are described in the general chapters of the documentation.
 
+Customize role label
+^^^^^^^^^^^^^^^^^^^^
+By default, Patroni will set corresponding labels on the pod it runs in based on node's role, such as ``role=master``.
+The key and value of label can be customized by `kubernetes.role_label`, `kubernetes.leader_label_value`, `kubernetes.follower_label_value` and `kubernetes.standby_leader_label_value`.
+
+Note that if you migrate from default role labels to custom ones, you can reduce downtime by following migration steps:
+
+1. Add a temporary label using original role value for the pod with `kubernetes.tmp_role_label` (like ``tmp_role``). Once pods are restarted they will get following labels set by Patroni:
+
+  .. code:: YAML
+
+    labels:
+      cluster-name: foo
+      role: master
+      tmp_role: master
+
+2. After all pods have been updated, modify the service selector to select the temporary label.
+
+  .. code:: YAML
+
+    selector:
+      cluster-name: foo
+      tmp_role: master
+
+3. Add your custom role label (e.g., set `kubernetes.leader_label_value=primary`). Once pods are restarted they will get following new labels set by Patroni:
+
+  .. code:: YAML
+
+    labels:
+      cluster-name: foo
+      role: primary
+      tmp_role: master
+
+4. After all pods have been updated again, modify the service selector to use new role value.
+
+  .. code:: YAML
+
+    selector:
+      cluster-name: foo
+      role: primary
+
+5. Finally, remove the temporary label from your configuration and update all pods.
+
+  .. code:: YAML
+
+    labels:
+      cluster-name: foo
+      role: primary
+
 Examples
 --------
 

docs/yaml_configuration.rst

@@ -155,7 +155,11 @@ Kubernetes
 -  **namespace**: (optional) Kubernetes namespace where Patroni pod is running. Default value is `default`.
 -  **labels**: Labels in format ``{label1: value1, label2: value2}``. These labels will be used to find existing objects (Pods and either Endpoints or ConfigMaps) associated with the current cluster. Also Patroni will set them on every object (Endpoint or ConfigMap) it creates.
 -  **scope\_label**: (optional) name of the label containing cluster name. Default value is `cluster-name`.
--  **role\_label**: (optional) name of the label containing role (master or replica). Patroni will set this label on the pod it runs in. Default value is ``role``.
+-  **role\_label**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``.
+-  **leader\_label\_value**: (optional) value of the pod label when Postgres role is ``master``. Default value is ``master``.
+-  **follower\_label\_value**: (optional) value of the pod label when Postgres role is ``replica``. Default value is ``replica``.
+-  **standby\_leader\_label\_value**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``.
+-  **tmp_\role\_label**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary.
 -  **use\_endpoints**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state.
 -  **pod\_ip**: (optional) IP address of the pod Patroni is running in. This value is required when `use_endpoints` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted.
 -  **ports**: (optional) if the Service object has the name for the port, the same name must appear in the Endpoint object, otherwise service won't work. For example, if your service is defined as ``{Kind: Service, spec: {ports: [{name: postgresql, port: 5432, targetPort: 5432}]}}``, then you have to set ``kubernetes.ports: [{"name": "postgresql", "port": 5432}]`` and Patroni will use it for updating subsets of the leader Endpoint. This parameter is used only if `kubernetes.use_endpoints` is set.
@@ -335,17 +339,27 @@ Here is an example of both **http_extra_headers** and **https_extra_headers**:
           https_extra_headers:
             'Strict-Transport-Security': 'max-age=31536000; includeSubDomains'
 
+.. warning::
+
+    - The ``restapi.connect_address`` must be accessible from all nodes of a given Patroni cluster. Internally Patroni is using it during the leader race to find nodes with minimal replication lag.
+    - If you enabled client certificates validation (``restapi.verify_client`` is set to ``required``), you also **must** provide **valid client certificates** in the ``ctl.certfile``, ``ctl.keyfile``, ``ctl.keyfile_password``. If not provided, Patroni will not work correctly.
+
+
 .. _patronictl_settings:
 
 CTL
 ---
 -  **ctl**: (optional)
 
+   -  **authentication**:
+
+      -  **username**: Basic-auth username for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "username" parameter.
+      -  **password**: Basic-auth password for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "password" parameter.
    -  **insecure**: Allow connections to REST API without verifying SSL certs.
    -  **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter.
-   -  **certfile**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter.
-   -  **keyfile**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter.
-   -  **keyfile\_password**: Specifies a password for decrypting the keyfile. If not provided patronictl will use the value provided for REST API "keyfile\_password" parameter.
+   -  **certfile**: Specifies the file with the client certificate in the PEM format.
+   -  **keyfile**: Specifies the file with the client secret key in the PEM format.
+   -  **keyfile\_password**: Specifies a password for decrypting the client keyfile.
 
 Watchdog
 --------

features/basic_replication.feature

@@ -79,6 +79,7 @@ Feature: basic replication
     When I add the table buz to postgres2
     Then table buz is present on postgres0 after 20 seconds
 
+  @reject-duplicate-name
   Scenario: check graceful rejection when two nodes have the same name
     Given I start duplicate postgres0 on port 8011
     Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log

features/environment.py

@@ -59,7 +59,7 @@ class AbstractController(abc.ABC):
                 break
             time.sleep(1)
         else:
-            assert False,\
+            assert False, \
                 "{0} instance is not available for queries after {1} seconds".format(self._name, max_wait_limit)
 
     def stop(self, kill=False, timeout=15, _=False):
@@ -1082,7 +1082,9 @@ def before_all(context):
                            'PATRONI_RESTAPI_CERTFILE': context.certfile,
                            'PATRONI_RESTAPI_KEYFILE': context.keyfile,
                            'PATRONI_RESTAPI_VERIFY_CLIENT': 'required',
-                           'PATRONI_CTL_INSECURE': 'on'})
+                           'PATRONI_CTL_INSECURE': 'on',
+                           'PATRONI_CTL_CERTFILE': context.certfile,
+                           'PATRONI_CTL_KEYFILE': context.keyfile})
         ctl.update({'cacert': context.certfile, 'certfile': context.certfile, 'keyfile': context.keyfile})
     context.request_executor = PatroniRequest({'ctl': ctl}, True)
     context.dcs_ctl = context.pctl.known_dcs[context.pctl.dcs](context)
@@ -1144,3 +1146,5 @@ def before_scenario(context, scenario):
                 break
     if 'dcs-failsafe' in scenario.effective_tags and not context.dcs_ctl._handle:
         scenario.skip('it is not possible to control state of {0} from tests'.format(context.dcs_ctl.name()))
+    if 'reject-duplicate-name' in scenario.effective_tags and context.dcs_ctl.name() == 'raft':
+        scenario.skip('Flaky test with Raft')

features/steps/basic_replication.py

@@ -21,7 +21,7 @@ def start_duplicate_patroni(context, name, port):
         context.pctl.start('dup-' + name, custom_config=config)
         assert False, "Process was expected to fail"
     except AssertionError as e:
-        assert 'is not running after being started' in str(e),\
+        assert 'is not running after being started' in str(e), \
             "No error was raised by duplicate start of {0} ".format(name)
 
 
@@ -88,14 +88,14 @@ def table_is_present_on(context, table_name, pg_name, max_replication_delay):
             break
         sleep(1)
     else:
-        assert False,\
+        assert False, \
             "Table {0} is not present on {1} after {2} seconds".format(table_name, pg_name, max_replication_delay)
 
 
 @then('{pg_name:w} role is the {pg_role:w} after {max_promotion_timeout:d} seconds')
 def check_role(context, pg_name, pg_role, max_promotion_timeout):
     max_promotion_timeout *= context.timeout_multiplier
-    assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)),\
+    assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)), \
         "{0} role didn't change to {1} after {2} seconds".format(pg_name, pg_role, max_promotion_timeout)
 
 
@@ -111,5 +111,5 @@ def replication_works(context, primary, replica, time_limit):
 @then('there is a "{message}" {level:w} in the {node} patroni log')
 def check_patroni_log(context, message, level, node):
     messsages_of_level = context.pctl.read_patroni_log(node, level)
-    assert any(message in line for line in messsages_of_level),\
+    assert any(message in line for line in messsages_of_level), \
         "There was no {0} {1} in the {2} patroni log".format(message, level, node)

features/steps/citus.py

@@ -125,5 +125,5 @@ def check_transaction(context, name):
 
 @step("a transaction finishes in {timeout:d} seconds")
 def check_transaction_timeout(context, timeout):
-    assert (datetime.now(tzutc) - context.xact_start).seconds > timeout,\
+    assert (datetime.now(tzutc) - context.xact_start).seconds > timeout, \
         "a transaction finished earlier than in {0} seconds".format(timeout)

features/steps/patroni_api.py

@@ -98,7 +98,7 @@ def do_run(context, cmd):
 @then('I receive a response {component:w} {data}')
 def check_response(context, component, data):
     if component == 'code':
-        assert context.status_code == int(data),\
+        assert context.status_code == int(data), \
             "status code {0} != {1}, response: {2}".format(context.status_code, data, context.response)
     elif component == 'returncode':
         assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code,
@@ -158,7 +158,7 @@ def check_http_response(context, url, value, timeout, negate=False):
             break
         time.sleep(1)
     else:
-        assert False,\
+        assert False, \
             "Value {0} is {1} present in response after {2} seconds".format(value, "not" if not negate else "", timeout)
 
 

patroni/api.py

@@ -463,6 +463,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
         * ``patroni_dcs_last_seen``: epoch timestamp when DCS was last contacted successfully;
         * ``patroni_pending_restart``: ``1`` if this PostgreSQL node is pending a restart, else ``0``;
         * ``patroni_is_paused``: ``1`` if Patroni is in maintenance node, else ``0``.
+
+        For PostgreSQL v9.6+ the response will also have the following:
+
+        * ``patroni_postgres_streaming``: 1 if Postgres is streaming from another node, else ``0``;
+        * ``patroni_postgres_in_archive_recovery``: ``1`` if Postgres isn't streaming and
+                                                    there is ``restore_command`` available, else ``0``.
         """
         postgres = self.get_postgresql_status(True)
         patroni = self.server.patroni

patroni/config.py

@@ -329,9 +329,19 @@ class Config(object):
 
     @staticmethod
     def _process_postgresql_parameters(parameters: Dict[str, Any], is_local: bool = False) -> Dict[str, Any]:
-        return {name: value for name, value in (parameters or {}).items()
-                if name not in ConfigHandler.CMDLINE_OPTIONS
-                or not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)}
+        pg_params: Dict[str, Any] = {}
+
+        for name, value in (parameters or {}).items():
+            if name not in ConfigHandler.CMDLINE_OPTIONS:
+                pg_params[name] = value
+            elif not is_local:
+                if ConfigHandler.CMDLINE_OPTIONS[name][1](value):
+                    pg_params[name] = value
+                else:
+                    logging.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
+                                    name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])
+
+        return pg_params
 
     def _safe_copy_dynamic_configuration(self, dynamic_configuration: Dict[str, Any]) -> Dict[str, Any]:
         config = deepcopy(self.__DEFAULT_CONFIG)
@@ -452,9 +462,10 @@ class Config(object):
                     ret[param] = value
             return ret
 
-        restapi_auth = _get_auth('restapi')
-        if restapi_auth:
-            ret['restapi']['authentication'] = restapi_auth
+        for section in ('ctl', 'restapi'):
+            auth = _get_auth(section)
+            if auth:
+                ret[section]['authentication'] = auth
 
         authentication = {}
         for user_type in ('replication', 'superuser', 'rewind'):
@@ -474,7 +485,8 @@ class Config(object):
                               'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'SERVICE_CHECK_TLS_SERVER_NAME',
                               'SERVICE_TAGS', 'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL',
                               'POD_IP', 'PORTS', 'LABELS', 'BYPASS_API_SERVICE', 'RETRIABLE_HTTP_CODES', 'KEY_PASSWORD',
-                              'USE_SSL', 'SET_ACLS', 'GROUP', 'DATABASE') and name:
+                              'USE_SSL', 'SET_ACLS', 'GROUP', 'DATABASE', 'LEADER_LABEL_VALUE', 'FOLLOWER_LABEL_VALUE',
+                              'STANDBY_LEADER_LABEL_VALUE', 'TMP_ROLE_LABEL') and name:
                     value = os.environ.pop(param)
                     if name == 'CITUS':
                         if suffix == 'GROUP':
@@ -531,9 +543,10 @@ class Config(object):
             elif name not in config or name in ['watchdog']:
                 config[name] = deepcopy(value) if value else {}
 
-        # restapi server expects to get restapi.auth = 'username:password'
-        if 'restapi' in config and 'authentication' in config['restapi']:
-            config['restapi']['auth'] = '{username}:{password}'.format(**config['restapi']['authentication'])
+        # restapi server expects to get restapi.auth = 'username:password' and similarly for `ctl`
+        for section in ('ctl', 'restapi'):
+            if section in config and 'authentication' in config[section]:
+                config[section]['auth'] = '{username}:{password}'.format(**config[section]['authentication'])
 
         # special treatment for old config
 

patroni/ctl.py

@@ -254,7 +254,6 @@ arg_cluster_name = click.argument('cluster_name', required=False,
 option_default_citus_group = click.option('--group', required=False, type=int, help='Citus group',
                                           default=lambda: click.get_current_context().obj.get('citus', {}).get('group'))
 option_citus_group = click.option('--group', required=False, type=int, help='Citus group')
-option_insecure = click.option('-k', '--insecure', is_flag=True, help='Allow connections to SSL sites without certs')
 role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 'standby', 'any', 'master'])
 
 
@@ -262,7 +261,7 @@ role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 's
 @click.option('--config-file', '-c', help='Configuration file',
               envvar='PATRONICTL_CONFIG_FILE', default=CONFIG_FILE_PATH)
 @click.option('--dcs-url', '--dcs', '-d', 'dcs_url', help='The DCS connect url', envvar='DCS_URL')
-@option_insecure
+@click.option('-k', '--insecure', is_flag=True, help='Allow connections to SSL sites without certs')
 @click.pass_context
 def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: bool) -> None:
     """Entry point of ``patronictl`` utility.
@@ -1544,7 +1543,7 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str,
             logging.debug(member)
 
             lag = member.get('lag', '')
-            member.update(c=name, member=member['name'], group=g,
+            member.update(cluster=name, member=member['name'], group=g,
                           host=member.get('host', ''), tl=member.get('timeline', ''),
                           role=member['role'].replace('_', ' ').title(),
                           lag_in_mb=round(lag / 1024 / 1024) if isinstance(lag, int) else lag,

patroni/dcs/__init__.py

@@ -15,7 +15,9 @@ from collections import defaultdict
 from copy import deepcopy
 from random import randint
 from threading import Event, Lock
-from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING
+from types import ModuleType
+from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING, \
+    Type, Iterator
 from urllib.parse import urlparse, urlunparse, parse_qsl
 
 from ..exceptions import PatroniFatalException
@@ -85,38 +87,83 @@ def dcs_modules() -> List[str]:
         return [module_prefix + name for _, name, is_pkg in pkgutil.iter_modules([dcs_dirname]) if not is_pkg]
 
 
-def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS':
-    modules = dcs_modules()
+def iter_dcs_classes(
+        config: Optional[Union['Config', Dict[str, Any]]] = None
+) -> Iterator[Tuple[str, Type['AbstractDCS']]]:
+    """Attempt to import DCS modules that are present in the given configuration.
+
+    .. note::
+            If a module successfully imports we can assume that all its requirements are installed.
+
+    :param config: configuration information with possible DCS names as keys. If given, only attempt to import DCS
+                   modules defined in the configuration. Else, if ``None``, attempt to import any supported DCS module.
+
+    :yields: a tuple containing the module ``name`` and the imported DCS class object.
+    """
+    for mod_name in dcs_modules():
+        name = mod_name.rpartition('.')[2]
+        if config is None or name in config:
 
-    for module_name in modules:
-        name = module_name.split('.')[-1]
-        if name in config:  # we will try to import only modules which have configuration section in the config file
             try:
-                module = importlib.import_module(module_name)
-                for key, item in module.__dict__.items():  # iterate through the module content
-                    # try to find implementation of AbstractDCS interface, class name must match with module_name
-                    if key.lower() == name and inspect.isclass(item) and issubclass(item, AbstractDCS):
-                        # propagate some parameters
-                        config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait',
-                                             'patronictl', 'ttl', 'retry_timeout') if p in config})
-                        # From citus section we only need "group" parameter, but will propagate everything just in case.
-                        if isinstance(config.get('citus'), dict):
-                            config[name].update(config['citus'])
-                        return item(config[name])
-            except ImportError:
-                logger.debug('Failed to import %s', module_name)
+                module = importlib.import_module(mod_name)
+                dcs_module = find_dcs_class_in_module(module)
+                if dcs_module:
+                    yield name, dcs_module
 
-    available_implementations: List[str] = []
-    for module_name in modules:
-        name = module_name.split('.')[-1]
-        try:
-            module = importlib.import_module(module_name)
-            available_implementations.extend(name for key, item in module.__dict__.items() if key.lower() == name
-                                             and inspect.isclass(item) and issubclass(item, AbstractDCS))
-        except ImportError:
-            logger.info('Failed to import %s', module_name)
-    raise PatroniFatalException("""Can not find suitable configuration of distributed configuration store
-Available implementations: """ + ', '.join(sorted(set(available_implementations))))
+            except ImportError:
+                logger.log(logging.DEBUG if config is not None else logging.INFO,
+                           'Failed to import %s', mod_name)
+
+
+def find_dcs_class_in_module(module: ModuleType) -> Optional[Type['AbstractDCS']]:
+    """Try to find the implementation of :class:`AbstractDCS` interface in *module* matching the *module* name.
+
+    :param module: Imported DCS module.
+
+    :returns: class with a name matching the name of *module* that implements :class:`AbstractDCS` or ``None`` if not
+              found.
+    """
+    module_name = module.__name__.rpartition('.')[2]
+    return next(
+        (obj for obj_name, obj in module.__dict__.items()
+         if (obj_name.lower() == module_name
+             and inspect.isclass(obj) and issubclass(obj, AbstractDCS))),
+        None)
+
+
+def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS':
+    """Attempt to load a Distributed Configuration Store from known available implementations.
+
+    .. note::
+        Using the list of available DCS modules returned by :func:`iter_dcs_modules` attempt to dynamically import and
+        instantiate the class that implements a DCS using the abstract class :class:`AbstractDCS`.
+
+        Basic top-level configuration parameters retrieved from *config* are propagated to the DCS specific config
+        before being passed to the module DCS class.
+
+        If no module is found to satisfy configuration then report and log an error. This will cause Patroni to exit.
+
+    :raises :exc:`PatroniFatalException`: if a load of all available DCS modules have been tried and none succeeded.
+
+    :param config: object or dictionary with Patroni configuration. This is normally a representation of the main
+                   Patroni
+
+    :returns: The first successfully loaded DCS module which is an implementation of :class:`AbstractDCS`.
+    """
+    for name, dcs_class in iter_dcs_classes(config):
+        # Propagate some parameters from top level of config if defined to the DCS specific config section.
+        config[name].update({
+            p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait',
+                                   'patronictl', 'ttl', 'retry_timeout')
+            if p in config})
+        # From citus section we only need "group" parameter, but will propagate everything just in case.
+        if isinstance(config.get('citus'), dict):
+            config[name].update(config['citus'])
+        return dcs_class(config[name])
+
+    raise PatroniFatalException(
+        f"Can not find suitable configuration of distributed configuration store\n"
+        f"Available implementations: {', '.join(sorted([n for n, _ in iter_dcs_classes()]))}")
 
 
 _Version = Union[int, str]
@@ -508,36 +555,50 @@ class TimelineHistory(NamedTuple):
         return TimelineHistory(version, value, lines)
 
 
-class Cluster(NamedTuple):
-    """Immutable object (namedtuple) which represents PostgreSQL cluster.
+class Cluster(NamedTuple('Cluster',
+                         [('initialize', Optional[str]),
+                          ('config', Optional[ClusterConfig]),
+                          ('leader', Optional[Leader]),
+                          ('last_lsn', int),
+                          ('members', List[Member]),
+                          ('failover', Optional[Failover]),
+                          ('sync', SyncState),
+                          ('history', Optional[TimelineHistory]),
+                          ('slots', Optional[Dict[str, int]]),
+                          ('failsafe', Optional[Dict[str, str]]),
+                          ('workers', Dict[int, 'Cluster'])])):
+    """Immutable object (namedtuple) which represents PostgreSQL or Citus cluster.
+
+    .. note::
+        We are using an old-style attribute declaration here because otherwise it is not possible to override `__new__`
+        method. Without it the *workers* by default gets always the same :class:`dict` object that could be mutated.
+
     Consists of the following fields:
-    :param initialize: shows whether this cluster has initialization key stored in DC or not.
-    :param config: global dynamic configuration, reference to `ClusterConfig` object
-    :param leader: `Leader` object which represents current leader of the cluster
-    :param last_lsn: int or long object containing position of last known leader LSN.
-        This value is stored in the `/status` key or `/optime/leader` (legacy) key
-    :param members: list of Member object, all PostgreSQL cluster members including leader
-    :param failover: reference to `Failover` object
-    :param sync: reference to `SyncState` object, last observed synchronous replication state.
-    :param history: reference to `TimelineHistory` object
-    :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}
-    :param failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list.
-    :param workers: workers of the Citus cluster, optional. Format: {int(group): Cluster()}
+
+    :ivar initialize: shows whether this cluster has initialization key stored in DC or not.
+    :ivar config: global dynamic configuration, reference to `ClusterConfig` object.
+    :ivar leader: :class:`Leader` object which represents current leader of the cluster.
+    :ivar last_lsn: :class:int object containing position of last known leader LSN.
+                     This value is stored in the `/status` key or `/optime/leader` (legacy) key.
+    :ivar members: list of:class:` Member` objects, all PostgreSQL cluster members including leader
+    :ivar failover: reference to :class:`Failover` object.
+    :ivar sync: reference to :class:`SyncState` object, last observed synchronous replication state.
+    :ivar history: reference to `TimelineHistory` object.
+    :ivar slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}.
+    :ivar failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list.
+    :ivar workers: dictionary of workers of the Citus cluster, optional. Each key is an :class:`int` representing
+                   the group, and the corresponding value is a :class:`Cluster` instance.
     """
-    initialize: Optional[str]
-    config: Optional[ClusterConfig]
-    leader: Optional[Leader]
-    last_lsn: int
-    members: List[Member]
-    failover: Optional[Failover]
-    sync: SyncState
-    history: Optional[TimelineHistory]
-    slots: Optional[Dict[str, int]]
-    failsafe: Optional[Dict[str, str]]
-    workers: Dict[int, 'Cluster'] = {}
+
+    def __new__(cls, *args: Any, **kwargs: Any):
+        """Make workers argument optional and set it to an empty dict object."""
+        if len(args) < len(cls._fields) and 'workers' not in kwargs:
+            kwargs['workers'] = {}
+        return super(Cluster, cls).__new__(cls, *args, **kwargs)
 
     @staticmethod
     def empty() -> 'Cluster':
+        """Produce an empty :class:`Cluster` instance."""
         return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None)
 
     def is_empty(self):
@@ -586,24 +647,25 @@ class Cluster(NamedTuple):
 
     def get_replication_slots(self, my_name: str, role: str, nofailover: bool,
                               major_version: int, show_error: bool = False) -> Dict[str, Dict[str, Any]]:
-        # if the replicatefrom tag is set on the member - we should not create the replication slot for it on
-        # the current primary, because that member would replicate from elsewhere. We still create the slot if
-        # the replicatefrom destination member is currently not a member of the cluster (fallback to the
-        # primary), or if replicatefrom destination member happens to be the current primary
-        use_slots = self.use_slots
-        if role in ('master', 'primary', 'standby_leader'):
-            slot_members = [m.name for m in self.members if use_slots and m.name != my_name
-                            and (m.replicatefrom is None or m.replicatefrom == my_name
-                                 or not self.has_member(m.replicatefrom))]
-            permanent_slots = self.__permanent_slots if use_slots and \
-                role in ('master', 'primary') else self.__permanent_physical_slots
-        else:
-            # only manage slots for replicas that replicate from this one, except for the leader among them
-            slot_members = [m.name for m in self.members if use_slots
-                            and m.replicatefrom == my_name and m.name != self.leader_name]
-            permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {}
+        """Lookup configured slot names in the DCS, report issues found and merge with permanent slots.
 
-        slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members}
+        Will log an error if:
+
+            * Conflicting slot names between members are found
+            * Any logical slots are disabled, due to version compatibility, and *show_error* is ``True``.
+
+        :param my_name: name of this node.
+        :param role: role of this node.
+        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param major_version: postgresql major version.
+        :param show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found.
+
+        :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
+        """
+        slot_members: List[str] = self._get_slot_members(my_name, role) if self.use_slots else []
+
+        slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'}
+                                            for name in slot_members}
 
         if len(slots) < len(slot_members):
             # Find which names are conflicting for a nicer error message
@@ -611,11 +673,38 @@ class Cluster(NamedTuple):
             for name in slot_members:
                 slot_conflicts[slot_name_from_member_name(name)].append(name)
             logger.error("Following cluster members share a replication slot name: %s",
-                         "; ".join("{} map to {}".format(", ".join(v), k)
+                         "; ".join(f"{', '.join(v)} map to {k}"
                                    for k, v in slot_conflicts.items() if len(v) > 1))
 
-        # "merge" replication slots for members with permanent_replication_slots
+        permanent_slots: dict[str, Any] = self._get_permanent_slots(role, nofailover) if self.use_slots else {}
+        disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
+            slots, permanent_slots, my_name, major_version)
+
+        if disabled_permanent_logical_slots and show_error:
+            logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. "
+                         "Following slots will not be created: %s.", disabled_permanent_logical_slots)
+
+        return slots
+
+    @staticmethod
+    def _merge_permanent_slots(slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str,
+                               major_version: int) -> List[str]:
+        """Merge replication *slots* for members with *permanent_slots*.
+
+        Perform validation of configured permanent slot name, skipping invalid names.
+
+        Will update *slots* in-line based on ``type`` of slot, ``physical`` or ``logical``, and name of node.
+        Type is assumed to be ``physical`` if there are no attributes stored as the slot value.
+
+        :param slots: Slot names with existing attributes if known.
+        :param my_name: name of this node.
+        :param permanent_slots: dictionary containing slot name key and slot information values.
+        :param major_version: postgresql major version.
+
+        :returns: List of disabled permanent, logical slot names, if postgresql version < 11.
+        """
         disabled_permanent_logical_slots: List[str] = []
+
         for name, value in permanent_slots.items():
             if not slot_name_re.match(name):
                 logger.error("Invalid permanent replication slot name '%s'", name)
@@ -632,7 +721,8 @@ class Cluster(NamedTuple):
                     if name != slot_name_from_member_name(my_name):
                         slots[name] = value
                     continue
-                elif value['type'] == 'logical' and value.get('database') and value.get('plugin'):
+
+                if value['type'] == 'logical' and value.get('database') and value.get('plugin'):
                     if major_version < 110000:
                         disabled_permanent_logical_slots.append(name)
                     elif name in slots:
@@ -643,12 +733,62 @@ class Cluster(NamedTuple):
                     continue
 
             logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
+        return disabled_permanent_logical_slots
 
-        if disabled_permanent_logical_slots and show_error:
-            logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. "
-                         "Following slots will not be created: %s.", disabled_permanent_logical_slots)
+    def _get_permanent_slots(self, role: str, nofailover: bool) -> Dict[str, Any]:
+        """Get configured permanent slot names.
 
-        return slots
+        .. note::
+            Permanent logical replication slots are only considered if ``use_slots`` configuration is enabled. Also,
+            only considered if *role* is ``primary`` or if it is a promotable ``replica`` -- what excludes a
+            ``standby_leader`` or ``replica`` with ``nofailover`` tag enabled. That combination is used for failing
+            over logical replication slots, and the latter nodes are not eligible for such task.
+
+            Permanent physical slots are only considered if *role* is ``primary`` or ``standby_leader``, independently
+            if ``use_slots`` is enabled or not. That is done that way because even if Patroni itself is not using slots
+            to replicate among its members when ``use_slots`` is disabled, the user may still have configured Patroni to
+            keep permanent physical slots used out of Patroni.
+
+        :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``.
+                     or logical slots being consumed.
+        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+
+        :returns: dictionary of permanent slot names mapped to attributes.
+        """
+        if role in ('master', 'primary', 'standby_leader'):
+            permanent_slots = (self.__permanent_slots
+                               if role in ('master', 'primary')
+                               else self.__permanent_physical_slots)
+        else:
+            permanent_slots = self.__permanent_logical_slots if not nofailover else {}
+        return permanent_slots
+
+    def _get_slot_members(self, my_name: str, role: str) -> List[str]:
+        """Get a list of member names that have replication slots sourcing from this node.
+
+        If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on
+        the current primary, because that member would replicate from elsewhere. We still create the slot if
+        the ``replicatefrom`` destination member is currently not a member of the cluster (fallback to the
+        primary), or if ``replicatefrom`` destination member happens to be the current primary.
+
+        :param my_name: name of this node.
+        :param role: role of this node, if this is a ``primary`` or ``standby_leader`` return list of members
+                     replicating from this node. If not then return a list of members replicating as cascaded
+                     replicas from this node.
+
+        :returns: list of member names.
+        """
+        if role in ('master', 'primary', 'standby_leader'):
+            slot_members = [m.name for m in self.members
+                            if m.name != my_name
+                            and (m.replicatefrom is None
+                                 or m.replicatefrom == my_name
+                                 or not self.has_member(m.replicatefrom))]
+        else:
+            # only manage slots for replicas that replicate from this one, except for the leader among them
+            slot_members = [m.name for m in self.members
+                            if m.replicatefrom == my_name and m.name != self.leader_name]
+        return slot_members
 
     def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool:
         if major_version < 110000:
@@ -907,7 +1047,9 @@ class AbstractDCS(abc.ABC):
             raise
 
         self._last_seen = int(time.time())
-        self._last_status = {self._OPTIME: cluster.last_lsn, 'slots': cluster.slots}
+        self._last_status = {self._OPTIME: cluster.last_lsn}
+        if cluster.slots:
+            self._last_status['slots'] = cluster.slots
         self._last_failsafe = cluster.failsafe
 
         with self._cluster_thread_lock:

patroni/dcs/consul.py

@@ -15,7 +15,7 @@ from urllib3.exceptions import HTTPError
 from urllib.parse import urlencode, urlparse, quote
 from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING
 
-from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
     TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
 from ..exceptions import DCSError
 from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT

patroni/dcs/etcd.py

@@ -21,7 +21,7 @@ from urllib.parse import urlparse
 from urllib3 import Timeout
 from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError
 
-from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
     TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
 from ..exceptions import DCSError
 from ..request import get as requests_get

patroni/dcs/etcd3.py

@@ -15,7 +15,7 @@ from urllib3.exceptions import ReadTimeoutError, ProtocolError
 from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
 
-from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\
+from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, \
     TimelineHistory, catch_return_false_exception, citus_group_re
 from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry
 from ..exceptions import DCSError, PatroniException
@@ -630,6 +630,16 @@ class PatroniEtcd3Client(Etcd3Client):
 
         return ret
 
+    def txn(self, compare: Dict[str, Any], success: Dict[str, Any],
+            failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
+        ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry)
+        # Here we abuse the fact that the `failure` is only set in the call from update_leader().
+        # In all other cases the txn() call failure may be an indicator of a stale cache,
+        # and therefore we want to restart watcher.
+        if not failure and not ret:
+            self._restart_watcher()
+        return ret
+
 
 class Etcd3(AbstractEtcd):
 

patroni/dcs/kubernetes.py

@@ -19,10 +19,10 @@ from urllib3.exceptions import HTTPError
 from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING
 
-from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
     TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re
 from ..exceptions import DCSError
-from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\
+from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \
     Retry, RetryFailedError, tzutc, uri, USER_AGENT
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
@@ -752,6 +752,10 @@ class Kubernetes(AbstractDCS):
         self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items())
         self._namespace = config.get('namespace') or 'default'
         self._role_label = config.get('role_label', 'role')
+        self._leader_label_value = config.get('leader_label_value', 'master')
+        self._follower_label_value = config.get('follower_label_value', 'replica')
+        self._standby_leader_label_value = config.get('standby_leader_label_value', 'standby-leader')
+        self._tmp_role_label = config.get('tmp_role_label')
         self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME
         super(Kubernetes, self).__init__({**config, 'namespace': ''})
         if self._citus_group:
@@ -1263,19 +1267,30 @@ class Kubernetes(AbstractDCS):
     def touch_member(self, data: Dict[str, Any]) -> bool:
         cluster = self.cluster
         if cluster and cluster.leader and cluster.leader.name == self._name:
-            role = 'master'
+            role = self._leader_label_value
+            tmp_role = 'master'
         elif data['state'] == 'running' and data['role'] not in ('master', 'primary'):
-            role = data['role']
+            role = {
+                'replica': self._follower_label_value,
+                'standby-leader': self._standby_leader_label_value,
+            }.get(data['role'], data['role'])
+            tmp_role = data['role']
         else:
             role = None
+            tmp_role = None
+
+        role_labels = {self._role_label: role}
+        if self._tmp_role_label:
+            role_labels[self._tmp_role_label] = tmp_role
 
         member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
         pod_labels = member and member.data.pop('pod_labels', None)
         ret = member and pod_labels is not None\
-            and pod_labels.get(self._role_label) == role and deep_compare(data, member.data)
+            and all(pod_labels.get(k) == v for k, v in role_labels.items())\
+            and deep_compare(data, member.data)
 
         if not ret:
-            metadata = {'namespace': self._namespace, 'name': self._name, 'labels': {self._role_label: role},
+            metadata = {'namespace': self._namespace, 'name': self._name, 'labels': role_labels,
                         'annotations': {'status': json.dumps(data, separators=(',', ':'))}}
             body = k8s_client.V1Pod(metadata=k8s_client.V1ObjectMeta(**metadata))
             ret = self._api.patch_namespaced_pod(self._name, self._namespace, body)

patroni/ha.py

@@ -56,7 +56,7 @@ class _MemberStatus(NamedTuple):
         # If one of those is not in a response we want to count the node as not healthy/reachable
         wal: Dict[str, Any] = json.get('wal') or json['xlog']
         # abuse difference in primary/replica response format
-        in_recovery = not bool(wal.get('location')) or json.get('role') in ('master', 'primary')
+        in_recovery = not (bool(wal.get('location')) or json.get('role') in ('master', 'primary'))
         timeline = json.get('timeline', 0)
         dcs_last_seen = json.get('dcs_last_seen', 0)
         lsn = int(in_recovery and max(wal.get('received_location', 0), wal.get('replayed_location', 0)))
@@ -473,7 +473,8 @@ class Ha(object):
 
         # timeout > 0 indicates that we still have the leader lock, and it was just updated
         if timeout\
-                and data.get('Database cluster state') in ('in production', 'shutting down', 'shut down')\
+                and data.get('Database cluster state') in ('in production', 'in crash recovery',
+                                                           'shutting down', 'shut down')\
                 and self.state_handler.state == 'crashed'\
                 and self.state_handler.role in ('primary', 'master')\
                 and not self.state_handler.config.recovery_conf_exists():
@@ -496,6 +497,7 @@ class Ha(object):
 
         role = 'replica'
         if self.has_lock() and not self.is_standby_cluster():
+            self._rewind.reset_state()  # we want to later trigger CHECKPOINT after promote
             msg = "starting as readonly because i had the session lock"
             node_to_follow = None
         else:
@@ -525,10 +527,17 @@ class Ha(object):
         return msg
 
     def _get_node_to_follow(self, cluster: Cluster) -> Union[Leader, Member, None]:
-        # determine the node to follow. If replicatefrom tag is set,
-        # try to follow the node mentioned there, otherwise, follow the leader.
-        if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)):
+        """Determine the node to follow.
+
+        :param cluster: the currently known cluster state from DCS.
+
+        :returns: the node which we should be replicating from.
+        """
+        # The standby leader or when there is no standby leader we want to follow
+        # the remote member, except when there is no standby leader in pause.
+        if self.is_standby_cluster() and (self.has_lock(False) or self.cluster.is_unlocked() and not self.is_paused()):
             node_to_follow = self.get_remote_member()
+        # If replicatefrom tag is set, try to follow the node mentioned there, otherwise, follow the leader.
         elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
             node_to_follow = cluster.get_member(self.patroni.replicatefrom)
         else:
@@ -768,18 +777,14 @@ class Ha(object):
                 self.state_handler.sync_handler.set_synchronous_standby_names(
                     CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet())
             if self.state_handler.role not in ('master', 'promoted', 'primary'):
-                def on_success():
-                    self._rewind.reset_state()
-                    logger.info("cleared rewind state after becoming the leader")
-
                 def before_promote():
                     self.notify_citus_coordinator('before_promote')
 
                 with self._async_response:
                     self._async_response.reset()
+
                 self._async_executor.try_run_async('promote', self.state_handler.promote,
-                                                   args=(self.dcs.loop_wait, self._async_response,
-                                                         before_promote, on_success))
+                                                   args=(self.dcs.loop_wait, self._async_response, before_promote))
             return promote_message
 
     def fetch_node_status(self, member: Member) -> _MemberStatus:
@@ -998,9 +1003,22 @@ class Ha(object):
                 return ret
 
         if self.state_handler.is_leader():
-            # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
-            return not self.is_paused() or not self.cluster.initialize\
-                or self.state_handler.sysid == self.cluster.initialize
+            if self.is_paused():
+                # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
+                return not self.cluster.initialize or self.state_handler.sysid == self.cluster.initialize
+
+            # We want to protect from the following scenario:
+            # 1. node1 is stressed so much that heart-beat isn't running regularly and the leader lock expires.
+            # 2. node2 promotes, gets heavy load and the situation described in 1 repeats.
+            # 3. Patroni on node1 comes back, notices that Postgres is running as primary but there is
+            #    no leader key and "happily" acquires the leader lock.
+            # That is, node1 discarded promotion of node2. To avoid it we want to detect timeline change.
+            my_timeline = self.state_handler.get_primary_timeline()
+            if my_timeline < self.cluster.timeline:
+                logger.warning('My timeline %s is behind last known cluster timeline %s',
+                               my_timeline, self.cluster.timeline)
+                return False
+            return True
 
         if self.is_paused():
             return False
@@ -1084,7 +1102,9 @@ class Ha(object):
             # 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):
+
+            # for a manual failover/switchover with a candidate, we should check the requested candidate only
+            if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location):
                 self.state_handler.set_role('demoted')
                 with self._async_executor:
                     self.release_leader_key_voluntarily(checkpoint_location)
@@ -1188,15 +1208,12 @@ class Ha(object):
                     logger.warning('Failover is possible only to a specific candidate in a paused state')
                 else:
                     if self.is_synchronous_mode():
-                        if failover.candidate and not self.cluster.sync.matches(failover.candidate):
+                        members = self.get_failover_candidates(check_sync=True)
+                        if failover.candidate and not members:
                             logger.warning('Failover candidate=%s does not match with sync_standbys=%s',
                                            failover.candidate, self.cluster.sync.sync_standby)
-                            members = []
-                        else:
-                            members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)]
                     else:
-                        members = [m for m in self.cluster.members
-                                   if not failover.candidate or m.name == failover.candidate]
+                        members = self.get_failover_candidates()
                     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'
@@ -1614,6 +1631,9 @@ class Ha(object):
             else:
                 if self._was_paused:
                     self.state_handler.schedule_sanity_checks_after_pause()
+                    # during pause people could manually do something with Postgres, therefore we want
+                    # to double check rewind conditions on replicas and maybe run CHECKPOINT on the primary
+                    self._rewind.reset_state()
                 self._was_paused = False
 
             if not self.cluster.has_member(self.state_handler.name):
@@ -1848,7 +1868,9 @@ class Ha(object):
                     # 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):
+
+                    # for a manual failover/switchover with a candidate, we should check the requested candidate only
+                    if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location):
                         self.dcs.delete_leader(checkpoint_location)
                         status['deleted'] = True
                     else:
@@ -1908,3 +1930,24 @@ class Ha(object):
 
         name = member.name if member else 'remote_member:{}'.format(uuid.uuid1())
         return RemoteMember.from_name_and_data(name, data)
+
+    def get_failover_candidates(self, check_sync: bool = False) -> List[Member]:
+        """Return list of candidates for either manual or automatic failover.
+
+        Mainly used to later be passed to ``Ha.is_failover_possible()``.
+
+        :param check_sync: if ``True``, also check against the sync key members
+
+        :returns: a list of ``Member`` ojects or an empty list if there is no candidate available
+        """
+        failover = self.cluster.failover
+        if check_sync:
+            # TODO: allow manual failover (=no leader specified) to async node
+            # every sync_standby or the candidate specified if is in sync_standbys
+            return [m for m in self.cluster.members
+                    if self.cluster.sync.matches(m.name)
+                    and (not failover or not failover.candidate or m.name == failover.candidate)]
+        else:
+            # every member or the candidate specified
+            return [m for m in self.cluster.members
+                    if not failover or not failover.candidate or m.name == failover.candidate]

patroni/postgresql/__init__.py

@@ -1125,8 +1125,8 @@ class Postgresql(object):
         except Exception as e:
             logger.error('Exception when calling `%s`: %r', cmd, e)
 
-    def promote(self, wait_seconds: int, task: CriticalTask, before_promote: Optional[Callable[..., Any]] = None,
-                on_success: Optional[Callable[..., Any]] = None) -> Optional[bool]:
+    def promote(self, wait_seconds: int, task: CriticalTask,
+                before_promote: Optional[Callable[..., Any]] = None) -> Optional[bool]:
         if self.role in ('promoted', 'master', 'primary'):
             return True
 
@@ -1152,8 +1152,6 @@ class Postgresql(object):
         ret = self.pg_ctl('promote', '-W')
         if ret:
             self.set_role('promoted')
-            if on_success is not None:
-                on_success()
             self.call_nowait(CallbackAction.ON_ROLE_CHANGE)
             ret = self._wait_promote(wait_seconds)
         return ret

patroni/postgresql/config.py

@@ -15,7 +15,7 @@ from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
-from ..validator import IntValidator
+from ..validator import IntValidator, EnumValidator
 
 if TYPE_CHECKING:  # pragma: no cover
     from . import Postgresql
@@ -258,14 +258,14 @@ def _false_validator(value: Any) -> bool:
     return False
 
 
-def _wal_level_validator(value: Any) -> bool:
-    return str(value).lower() in ('hot_standby', 'replica', 'logical')
-
-
 def _bool_validator(value: Any) -> bool:
     return parse_bool(value) is not None
 
 
+def _bool_is_true_validator(value: Any) -> bool:
+    return parse_bool(value) is True
+
+
 class ConfigHandler(object):
 
     # List of parameters which must be always passed to postmaster as command line options
@@ -286,8 +286,8 @@ class ConfigHandler(object):
         'listen_addresses': (None, _false_validator, 90100),
         'port': (None, _false_validator, 90100),
         'cluster_name': (None, _false_validator, 90500),
-        'wal_level': ('hot_standby', _wal_level_validator, 90100),
-        'hot_standby': ('on', _false_validator, 90100),
+        'wal_level': ('hot_standby', EnumValidator(('hot_standby', 'replica', 'logical')), 90100),
+        'hot_standby': ('on', _bool_is_true_validator, 90100),
         'max_connections': (100, IntValidator(min=25), 90100),
         'max_wal_senders': (10, IntValidator(min=3), 90100),
         'wal_keep_segments': (8, IntValidator(min=1), 90100),
@@ -297,7 +297,7 @@ class ConfigHandler(object):
         'track_commit_timestamp': ('off', _bool_validator, 90500),
         'max_replication_slots': (10, IntValidator(min=4), 90400),
         'max_worker_processes': (8, IntValidator(min=2), 90400),
-        'wal_log_hints': ('on', _false_validator, 90400)
+        'wal_log_hints': ('on', _bool_is_true_validator, 90400)
     })
 
     _RECOVERY_PARAMETERS = CaseInsensitiveSet(recovery_parameters.keys())

patroni/postgresql/rewind.py

@@ -280,7 +280,7 @@ class Rewind(object):
         """After promote issue a CHECKPOINT from a new thread and asynchronously check the result.
         In case if CHECKPOINT failed, just check that timeline in pg_control was updated."""
 
-        if self._state == REWIND_STATUS.INITIAL and self._postgresql.is_leader():
+        if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_leader():
             with self._checkpoint_task_lock:
                 if self._checkpoint_task:
                     with self._checkpoint_task:
@@ -370,7 +370,7 @@ class Rewind(object):
                 # it is the author of archive_command, who is responsible
                 # for not overriding the WALs already present in archive
                 logger.info('Trying to archive %s: %s', wal, cmd)
-                if self._postgresql.cancellable.call(shlex.split(cmd)) == 0:
+                if self._postgresql.cancellable.call([cmd], shell=True) == 0:
                     new_name = os.path.join(status_dir, wal + '.done')
                     try:
                         shutil.move(old_name, new_name)

patroni/postgresql/slots.py

@@ -130,7 +130,7 @@ class SlotsHandler(object):
         self._postgresql = postgresql
         self._advance = None
         self._replication_slots: Dict[str, Dict[str, Any]] = {}  # already existing replication slots
-        self._unready_logical_slots: Dict[str, Optional[int]] = {}
+        self._logical_slots_processing_queue: Dict[str, Optional[int]] = {}
         self.pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
         self.schedule()
 
@@ -190,7 +190,8 @@ class SlotsHandler(object):
             self._replication_slots = replication_slots
             self._schedule_load_slots = False
             if self._force_readiness_check:
-                self._unready_logical_slots = {n: None for n, v in replication_slots.items() if v['type'] == 'logical'}
+                self._logical_slots_processing_queue = {n: None for n, v in replication_slots.items()
+                                                        if v['type'] == 'logical'}
                 self._force_readiness_check = False
 
     def ignore_replication_slot(self, cluster: Cluster, name: str) -> bool:
@@ -327,10 +328,10 @@ class SlotsHandler(object):
                 self._ensure_physical_slots(slots)
 
                 if self._postgresql.is_leader():
-                    self._unready_logical_slots.clear()
+                    self._logical_slots_processing_queue.clear()
                     self._ensure_logical_slots_primary(slots)
                 elif cluster.slots and slots:
-                    self.check_logical_slots_readiness(cluster, nofailover, replicatefrom)
+                    self.check_logical_slots_readiness(cluster, replicatefrom)
 
                     ret = self._ensure_logical_slots_replica(cluster, slots)
 
@@ -347,48 +348,100 @@ class SlotsHandler(object):
         with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
             yield cur
 
-    def check_logical_slots_readiness(self, cluster: Cluster, nofailover: bool, replicatefrom: Optional[str]) -> None:
+    def check_logical_slots_readiness(self, cluster: Cluster, replicatefrom: Optional[str]) -> bool:
+        """Determine whether all known logical slots are synchronised from the leader.
+
+         1) Retrieve the current ``catalog_xmin`` value for the physical slot from the cluster leader, and
+         2) using previously stored list of "unready" logical slots, those which have yet to be checked hence have no
+            stored slot attributes,
+         3) store logical slot ``catalog_xmin`` when the physical slot ``catalog_xmin`` becomes valid.
+
+         :param cluster: object containing stateful information for the cluster.
+         :param replicatefrom: name of the member that should be used to replicate from.
+
+         :returns: ``False`` if any issue while checking logical slots readiness, ``True`` otherwise.
+         """
         catalog_xmin = None
-        if self._unready_logical_slots and cluster.leader:
+        if self._logical_slots_processing_queue and cluster.leader:
             slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom)
             try:
                 with self._get_leader_connection_cursor(cluster.leader) as cur:
                     cur.execute("SELECT slot_name, catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
                                 " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)",
-                                ([n for n, v in self._unready_logical_slots.items() if v is None] + [slot_name],))
+                                ([n for n, v in self._logical_slots_processing_queue.items()
+                                  if v is None] + [slot_name],))
                     slots = {row[0]: row[1] for row in cur}
                     if slot_name not in slots:
-                        return logger.warning('Physical slot %s does not exist on the primary', slot_name)
+                        logger.warning('Physical slot %s does not exist on the primary', slot_name)
+                        return False
                     catalog_xmin = slots.pop(slot_name)
             except Exception as e:
-                return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
-            # Remember catalog_xmin of logical slots on the primary when catalog_xmin of
-            # the physical slot became valid. Logical slots on replica will be safe to use after
-            # promote when catalog_xmin of the physical slot overtakes these values.
-            if catalog_xmin is not None:
-                for name, value in slots.items():
-                    self._unready_logical_slots[name] = value
-            else:  # Replica isn't streaming or the hot_standby_feedback isn't enabled
-                try:
-                    cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean")
-                    row = cur.fetchone()
-                    if row and not row[0]:
-                        logger.error('Logical slot failover requires "hot_standby_feedback".'
-                                     ' Please check postgresql.auto.conf')
-                except Exception as e:
-                    logger.error('Failed to check the hot_standby_feedback setting: %r', e)
-                return  # since `catalog_xmin` isn't valid further checks don't make any sense
+                logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
+                return False
 
-        for name in list(self._unready_logical_slots):
-            value = self._replication_slots.get(name)
-            # The logical slot on a replica is safe to use when the physical replica slot on the primary:
-            # 1. has a nonzero/non-null catalog_xmin
-            # 2. has a catalog_xmin that is not newer (greater) than the catalog_xmin of any slot on the standby
-            # 3. overtook the catalog_xmin of remembered values of logical slots on the primary.
-            if not value or catalog_xmin is not None and\
-                    self._unready_logical_slots[name] <= catalog_xmin <= value['catalog_xmin']:
-                del self._unready_logical_slots[name]
-                if value:
+            if not self._update_pending_logical_slot_primary(slots, catalog_xmin):
+                return False  # since `catalog_xmin` isn't valid further checks don't make any sense
+
+        self._ready_logical_slots(catalog_xmin)
+        return True
+
+    def _update_pending_logical_slot_primary(self, slots: Dict[str, Any], catalog_xmin: Optional[int] = None) -> bool:
+        """Store pending logical slot information for ``catalog_xmin`` on the primary.
+
+        Remember ``catalog_xmin`` of logical slots on the primary when ``catalog_xmin`` of the physical slot became
+        valid. Logical slots on replica will be safe to use after promote when ``catalog_xmin`` of the physical slot
+        overtakes these values.
+
+        :param slots: dictionary of slot information from the primary
+        :param catalog_xmin: ``catalog_xmin`` of the physical slot used by this replica to stream changes from primary.
+
+        :returns: ``False`` if any issue was faced while processing, ``True`` otherwise.
+        """
+        if catalog_xmin is not None:
+            for name, value in slots.items():
+                self._logical_slots_processing_queue[name] = value
+            return True
+
+        # Replica isn't streaming or the hot_standby_feedback isn't enabled
+        try:
+            cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean")
+            row = cur.fetchone()
+            if row and not row[0]:
+                logger.error('Logical slot failover requires "hot_standby_feedback".'
+                             ' Please check postgresql.auto.conf')
+        except Exception as e:
+            logger.error('Failed to check the hot_standby_feedback setting: %r', e)
+        return False
+
+    def _ready_logical_slots(self, primary_physical_catalog_xmin: Optional[int] = None) -> None:
+        """Ready logical slots by comparing primary physical slot ``catalog_xmin`` to logical ``catalog_xmin``.
+
+        The logical slot on a replica is safe to use when the physical replica slot on the primary:
+
+            1. has a nonzero/non-null ``catalog_xmin`` represented by ``primary_physical_xmin``.
+            2. has a ``catalog_xmin`` that is not newer (greater) than the ``catalog_xmin`` of any slot on the standby
+            3. overtook the ``catalog_xmin`` of remembered values of logical slots on the primary.
+
+        :param primary_physical_catalog_xmin: is the value retrieved from ``pg_catalog.pg_get_replication_slots()`` for
+                                              the physical replication slot on the primary.
+        """
+        # Make a copy of processing queue keys as a list as the queue dictionary is modified inside the loop.
+        for name in list(self._logical_slots_processing_queue):
+            primary_logical_catalog_xmin = self._logical_slots_processing_queue[name]
+            standby_logical_slot = self._replication_slots.get(name, {})
+            standby_logical_catalog_xmin = standby_logical_slot.get('catalog_xmin', 0)
+            if TYPE_CHECKING:  # pragma: no cover
+                assert primary_logical_catalog_xmin is not None
+
+            if (
+                    not standby_logical_slot
+                    or primary_physical_catalog_xmin is not None
+                    and primary_logical_catalog_xmin <= primary_physical_catalog_xmin <= standby_logical_catalog_xmin
+            ):
+
+                del self._logical_slots_processing_queue[name]
+
+                if standby_logical_slot:
                     logger.info('Logical slot %s is safe to be used after a failover', name)
 
     def copy_logical_slots(self, cluster: Cluster, create_slots: List[str]) -> None:
@@ -433,7 +486,7 @@ class SlotsHandler(object):
                     shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
-                self._unready_logical_slots[name] = None
+                self._logical_slots_processing_queue[name] = None
             fsync_dir(self._postgresql.slots_handler.pg_replslot_dir)
             self._postgresql.start()
 
@@ -446,6 +499,6 @@ class SlotsHandler(object):
         if self._advance:
             self._advance.on_promote()
 
-        if self._unready_logical_slots:
+        if self._logical_slots_processing_queue:
             logger.warning('Logical replication slots that might be unsafe to use after promote: %s',
-                           set(self._unready_logical_slots))
+                           set(self._logical_slots_processing_queue))

patroni/postgresql/sync.py

@@ -153,6 +153,72 @@ def parse_sync_standby_names(value: str) -> _SSN:
     return _SSN(sync_type, has_star, num, members)
 
 
+class _Replica(NamedTuple):
+    """Class representing a single replica that is eligible to be synchronous.
+
+    Attributes are taken from ``pg_stat_replication`` view and respective ``Cluster.members``.
+
+    :ivar pid: PID of walsender process.
+    :ivar application_name: matches with the ``Member.name``.
+    :ivar sync_state: possible values are: ``async``, ``potential``, ``quorum``, and ``sync``.
+    :ivar lsn: ``write_lsn``, ``flush_lsn``, or ``replay_lsn``, depending on the value of ``synchronous_commit`` GUC.
+    :ivar nofailover: whether the corresponding member has ``nofailover`` tag set to ``True``.
+    """
+    pid: int
+    application_name: str
+    sync_state: str
+    lsn: int
+    nofailover: bool
+
+
+class _ReplicaList(List[_Replica]):
+    """A collection of :class:``_Replica`` objects.
+
+    Values are reverse ordered by ``_Replica.sync_state`` and ``_Replica.lsn``.
+    That is, first there will be replicas that have ``sync_state`` == ``sync``, even if they are not
+    the most up-to-date in term of write/flush/replay LSN. It helps to keep the result of chosing new
+    synchronous nodes consistent in case if a synchronous standby member is slowed down OR async node
+    is receiving changes faster than the sync member. Such cases would trigger sync standby member
+    swapping, but only if lag on this member is exceeding a threshold (``maximum_lag_on_syncnode``).
+
+    :ivar max_lsn: maximum value of ``_Replica.lsn`` among all values. In case if there is just one
+                   element in the list we take value of ``pg_current_wal_lsn()``.
+    """
+
+    def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None:
+        """Create :class:``_ReplicaList`` object.
+
+        :param postgresql: reference to :class:``Postgresql`` object.
+        :param cluster: currently known cluster state from DCS.
+        """
+        super().__init__()
+
+        # We want to prioritize candidates based on `write_lsn``, ``flush_lsn``, or ``replay_lsn``.
+        # Which column exactly to pick depends on the values of ``synchronous_commit`` GUC.
+        sort_col = {
+            'remote_apply': 'replay',
+            'remote_write': 'write'
+        }.get(postgresql.synchronous_commit(), 'flush') + '_lsn'
+
+        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
+        for row in postgresql.pg_stat_replication():
+            member = members.get(row['application_name'])
+
+            # We want to consider only rows from ``pg_stat_replication` that:
+            # 1. are known to be streaming (write/flush/replay LSN are not NULL).
+            # 2. can be mapped to a ``Member`` of the ``Cluster``:
+            #   a. ``Member`` doesn't have ``nosync`` tag set;
+            #   b. PostgreSQL on the member is known to be running and accepting client connections.
+            if member and row[sort_col] is not None and member.is_running and not member.tags.get('nosync', False):
+                self.append(_Replica(row['pid'], row['application_name'],
+                                     row['sync_state'], row[sort_col], bool(member.nofailover)))
+
+        # Prefer replicas that are in state ``sync`` and with higher values of ``write``/``flush``/``replay`` LSN.
+        self.sort(key=lambda r: (r.sync_state, r.lsn), reverse=True)
+
+        self.max_lsn = max(self, key=lambda x: x.lsn).lsn if len(self) > 1 else postgresql.last_operation()
+
+
 class SyncHandler(object):
     """Class responsible for working with the `synchronous_standby_names`.
 
@@ -201,6 +267,21 @@ BEGIN
 END;$$""")
         self._postgresql.reset_cluster_info_state(None)  # Reset internal cache to query fresh values
 
+    def _process_replica_readiness(self, cluster: Cluster, replica_list: _ReplicaList) -> None:
+        """Flags replicas as truly "synchronous" when they have caught up with ``_primary_flush_lsn``.
+
+        :param cluster: current cluster topology from DCS
+        :param replica_list: collection of replicas that we want to evaluate.
+        """
+        for replica in replica_list:
+            # if standby name is listed in the /sync key we can count it as synchronous, otherwise
+            # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
+            if replica.application_name not in self._ready_replicas\
+                    and replica.application_name in self._ssn_data.members\
+                    and (cluster.sync.matches(replica.application_name)
+                         or replica.sync_state == 'sync' and replica.lsn >= self._primary_flush_lsn):
+                self._ready_replicas[replica.application_name] = replica.pid
+
     def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
         """Finds best candidates to be the synchronous standbys.
 
@@ -218,31 +299,8 @@ END;$$""")
         """
         self._handle_synchronous_standby_names_change()
 
-        # Pick candidates based on who has higher replay/remote_write/flush lsn.
-        sort_col = {
-            'remote_apply': 'replay',
-            'remote_write': 'write'
-        }.get(self._postgresql.synchronous_commit(), 'flush') + '_lsn'
-
-        pg_stat_replication = [(r['pid'], r['application_name'], r['sync_state'], r[sort_col])
-                               for r in self._postgresql.pg_stat_replication()
-                               if r[sort_col] is not None]
-
-        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
-        replica_list: List[Tuple[int, str, str, int, bool]] = []
-        # pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync.
-        # That is, alphabetically they are in the reversed order of priority.
-        # Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result
-        # consistent in case if a synchronous standby member is slowed down OR async node receiving
-        # changes faster than the sync member (very rare but possible).
-        # Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold.
-        for pid, app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: r[2:4], reverse=True):
-            member = members.get(app_name)
-            if member and member.is_running and not member.tags.get('nosync', False):
-                replica_list.append((pid, member.name, sync_state, replica_lsn, bool(member.nofailover)))
-
-        max_lsn = max(replica_list, key=lambda x: x[3])[3]\
-            if len(replica_list) > 1 else self._postgresql.last_operation()
+        replica_list = _ReplicaList(self._postgresql, cluster)
+        self._process_replica_readiness(cluster, replica_list)
 
         if TYPE_CHECKING:  # pragma: no cover
             assert self._postgresql.global_config is not None
@@ -253,17 +311,11 @@ END;$$""")
         candidates = CaseInsensitiveSet()
         sync_nodes = CaseInsensitiveSet()
         # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
-        for pid, app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[4]):
-            # if standby name is listed in the /sync key we can count it as synchronous, otherwice
-            # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
-            if app_name not in self._ready_replicas and app_name in self._ssn_data.members and\
-                    (cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn):
-                self._ready_replicas[app_name] = pid
-
-            if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
-                candidates.add(app_name)
-                if sync_state == 'sync' and app_name in self._ready_replicas:
-                    sync_nodes.add(app_name)
+        for replica in sorted(replica_list, key=lambda x: x.nofailover):
+            if sync_node_maxlag <= 0 or replica_list.max_lsn - replica.lsn <= sync_node_maxlag:
+                candidates.add(replica.application_name)
+                if replica.sync_state == 'sync' and replica.application_name in self._ready_replicas:
+                    sync_nodes.add(replica.application_name)
             if len(candidates) >= sync_node_count:
                 break
 

patroni/request.py

@@ -11,6 +11,19 @@ from .dcs import Member
 from .utils import USER_AGENT
 
 
+class HTTPSConnectionPool(urllib3.HTTPSConnectionPool):
+
+    def _validate_conn(self, *args: Any, **kwargs: Any) -> None:
+        """Override parent method to silence warnings about requests without certificate verification enabled."""
+
+
+class PatroniPoolManager(urllib3.PoolManager):
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super(PatroniPoolManager, self).__init__(*args, **kwargs)
+        self.pool_classes_by_scheme = {'http': urllib3.HTTPConnectionPool, 'https': HTTPSConnectionPool}
+
+
 class PatroniRequest(object):
     """Wrapper for performing requests to Patroni's REST API.
 
@@ -28,22 +41,30 @@ class PatroniRequest(object):
             * If none of the above applies, then it falls back to ``False``.
         """
         self._insecure = insecure
-        self._pool = urllib3.PoolManager(num_pools=10, maxsize=10)
+        self._pool = PatroniPoolManager(num_pools=10, maxsize=10)
         self.reload_config(config)
 
     @staticmethod
-    def _get_cfg_value(config: Union[Config, Dict[str, Any]], name: str) -> Union[Any, None]:
-        """Get value of *name* setting in *config*.
-
-        .. note::
-            *name* key will be searched only under ``ctl`` and ``restapi`` sections, in that order.
+    def _get_ctl_value(config: Union[Config, Dict[str, Any]], name: str, default: Any = None) -> Optional[Any]:
+        """Get value of *name* setting from the ``ctl`` section of the *config*.
 
         :param config: Patroni YAML configuration.
         :param name: name of the setting value to be retrieved.
 
-        :returns: value of ``ctl -> *name*`` or ``restapi -> *name*``, if either is present, ``None`` otherwise.
+        :returns: value of ``ctl -> *name*`` if present, ``None`` otherwise.
         """
-        return config.get('ctl', {}).get(name) or config.get('restapi', {}).get(name)
+        return config.get('ctl', {}).get(name, default)
+
+    @staticmethod
+    def _get_restapi_value(config: Union[Config, Dict[str, Any]], name: str) -> Optional[Any]:
+        """Get value of *name* setting from the ``restapi`` section of the *config*.
+
+        :param config: Patroni YAML configuration.
+        :param name: name of the setting value to be retrieved.
+
+        :returns: value of ``restapi -> *name*`` if present, ``None`` otherwise.
+        """
+        return config.get('restapi', {}).get(name)
 
     def _apply_pool_param(self, param: str, value: Any) -> None:
         """Configure *param* as *value* in the request manager.
@@ -65,12 +86,11 @@ class PatroniRequest(object):
             * ``cert``: gets translated to ``certfile``
             * ``key``: gets translated to ``keyfile``
 
-            Will attempt to fetch the requested key first from ``ctl`` section, and fall back to ``restapi`` section
-            if the former is missing.
+            Will attempt to fetch the requested key first from ``ctl`` section.
 
-        :returns: value of ``ctl -> *name*file`` or ``restapi -> *name*file`` if either is present, ``None`` otherwise.
+        :returns: value of ``ctl -> *name*file`` if present, ``None`` otherwise.
         """
-        value = self._get_cfg_value(config, name + 'file')
+        value = self._get_ctl_value(config, name + 'file')
         self._apply_pool_param(name + '_file', value)
         return value
 
@@ -79,37 +99,39 @@ class PatroniRequest(object):
 
         Configure these HTTP headers for requests:
 
-        * ``authorization``: based on Patroni' REST API authentication config;
+        * ``authorization``: based on Patroni' CTL or REST API authentication config;
         * ``user-agent``: based on `patroni.utils.USER_AGENT`.
 
         Also configure SSL related settings for requests:
 
         * ``ca_certs`` is configured if ``ctl -> cacert`` or ``restapi -> cafile`` is available;
-        * ``cert``, ``key`` and ``key_password`` are configured if ``ctl -> certile`` or ``restapi -> certfile`` is
-            available.
+        * ``cert``, ``key`` and ``key_password`` are configured if ``ctl -> certfile`` is available.
 
         :param config: Patroni YAML configuration.
         """
-        # ``restapi -> auth`` is equivalent to ``restapi -> authentication -> username`` + ``:`` +
-        # ``restapi -> authentication -> password``
-        self._pool.headers = urllib3.make_headers(basic_auth=self._get_cfg_value(config, 'auth'), user_agent=USER_AGENT)
+        # ``ctl -> auth`` is equivalent to ``ctl -> authentication -> username`` + ``:`` +
+        # ``ctl -> authentication -> password``. And the same for ``restapi -> auth``
+        basic_auth = self._get_ctl_value(config, 'auth') or self._get_restapi_value(config, 'auth')
+        self._pool.headers = urllib3.make_headers(basic_auth=basic_auth, user_agent=USER_AGENT)
+        self._pool.connection_pool_kw['cert_reqs'] = 'CERT_REQUIRED'
+
+        insecure = self._insecure if isinstance(self._insecure, bool)\
+            else self._get_ctl_value(config, 'insecure', False)
 
-        insecure = self._insecure if isinstance(self._insecure, bool) else config.get('ctl', {}).get('insecure', False)
         if self._apply_ssl_file_param(config, 'cert'):
-            #  With client certificate the cert_reqs must be set to CERT_REQUIRED even if insecure option is used
-            self._pool.connection_pool_kw['cert_reqs'] = 'CERT_REQUIRED'
-            #  The assert_hostname = False helps to silence warnings
-            self._pool.connection_pool_kw['assert_hostname'] = False if insecure else None
+            if insecure:  # The assert_hostname = False helps to silence warnings
+                self._pool.connection_pool_kw['assert_hostname'] = False
 
             self._apply_ssl_file_param(config, 'key')
-
-            password = self._get_cfg_value(config, 'keyfile_password')
+            password = self._get_ctl_value(config, 'keyfile_password')
             self._apply_pool_param('key_password', password)
         else:
-            self._pool.connection_pool_kw['cert_reqs'] = 'CERT_NONE' if insecure else 'CERT_REQUIRED'
+            if insecure:  # Disable server certificate validation if requested
+                self._pool.connection_pool_kw['cert_reqs'] = 'CERT_NONE'
+            self._pool.connection_pool_kw.pop('assert_hostname', None)
             self._pool.connection_pool_kw.pop('key_file', None)
 
-        cacert = config.get('ctl', {}).get('cacert') or config.get('restapi', {}).get('cafile')
+        cacert = self._get_ctl_value(config, 'cacert') or self._get_restapi_value(config, 'cafile')
         self._apply_pool_param('ca_certs', cacert)
 
     def request(self, method: str, url: str, body: Optional[Any] = None,

patroni/validator.py

@@ -11,11 +11,12 @@ import shutil
 import socket
 import subprocess
 
-from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, TYPE_CHECKING
+from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING
 
-from .utils import parse_int, split_host_port, data_directory_is_empty
+from .collections import CaseInsensitiveSet
 from .dcs import dcs_modules
 from .exceptions import ConfigParseError
+from .utils import parse_int, split_host_port, data_directory_is_empty
 
 
 def data_directory_empty(data_dir: str) -> bool:
@@ -784,7 +785,7 @@ class IntValidator(object):
         self.base_unit = base_unit
         self.raise_assert = raise_assert
 
-    def __call__(self, value: Union[int, str]) -> bool:
+    def __call__(self, value: Any) -> bool:
         """Check if *value* is a valid integer and within the expected range.
 
         .. note::
@@ -802,6 +803,39 @@ class IntValidator(object):
         return ret
 
 
+class EnumValidator(object):
+    """Validate enum setting
+
+    :ivar allowed_values: a ``set`` or ``CaseInsensitiveSet`` object with allowed enum values.
+    :ivar raise_assert: if an ``assert`` call should be performed regarding expected type and valid range.
+    """
+
+    def __init__(self, allowed_values: Tuple[str, ...],
+                 case_sensitive: bool = False, raise_assert: bool = False) -> None:
+        """Create an :class:`EnumValidator` object with given allowed values.
+
+        :param allowed_values: a tuple with allowed enum values
+        :param case_sensitive: set to ``True`` to do case sensitive comparisons
+        :param raise_assert: if an ``assert`` call should be performed regarding expected values.
+        """
+        self.allowed_values = set(allowed_values) if case_sensitive else CaseInsensitiveSet(allowed_values)
+        self.raise_assert = raise_assert
+
+    def __call__(self, value: Any) -> bool:
+        """Check if provided *value* could be found within *allowed_values*.
+
+        .. note::
+            If ``raise_assert`` is ``True`` and *value* is not valid, then an ``AssertionError`` will be triggered.
+        :param value: value to be checked.
+        :returns: ``True`` if *value* could be found within *allowed_values*.
+        """
+        ret = isinstance(value, str) and value in self.allowed_values
+
+        if self.raise_assert:
+            assert_(ret)
+        return ret
+
+
 def validate_watchdog_mode(value: Any) -> None:
     """Validate ``watchdog.mode`` configuration option.
 
@@ -827,15 +861,44 @@ validate_etcd = {
         "srv": str,
         "srv_suffix": str,
         "url": str,
-        "proxy": str})
+        "proxy": str
+    }),
+    Optional("protocol"): str,
+    Optional("username"): str,
+    Optional("password"): str,
+    Optional("cacert"): str,
+    Optional("cert"): str,
+    Optional("key"): str
 }
 
 schema = Schema({
     "name": str,
     "scope": str,
+    Optional("ctl"): {
+        Optional("insecure"): bool,
+        Optional("cacert"): str,
+        Optional("certfile"): str,
+        Optional("keyfile"): str,
+        Optional("keyfile_password"): str
+    },
     "restapi": {
         "listen": validate_host_port_listen,
         "connect_address": validate_connect_address,
+        Optional("authentication"): {
+            "username": str,
+            "password": str
+        },
+        Optional("certfile"): str,
+        Optional("keyfile"): str,
+        Optional("keyfile_password"): str,
+        Optional("cafile"): str,
+        Optional("ciphers"): str,
+        Optional("verify_client"): EnumValidator(("none", "optional", "required"),
+                                                 case_sensitive=True, raise_assert=True),
+        Optional("allowlist"): [str],
+        Optional("allowlist_include_members"): bool,
+        Optional("http_extra_headers"): dict,
+        Optional("https_extra_headers"): dict,
         Optional("request_queue_size"): IntValidator(min=0, max=4096, raise_assert=True)
     },
     Optional("bootstrap"): {
@@ -843,15 +906,64 @@ schema = Schema({
             Optional("ttl"): int,
             Optional("loop_wait"): int,
             Optional("retry_timeout"): int,
-            Optional("maximum_lag_on_failover"): int
+            Optional("maximum_lag_on_failover"): int,
+            Optional("maximum_lag_on_syncnode"): int,
+            Optional("postgresql"): {
+                Optional("parameters"): {
+                    Optional("max_connections"): int,
+                    Optional("max_locks_per_transaction"): int,
+                    Optional("max_prepared_transactions"): int,
+                    Optional("max_replication_slots"): int,
+                    Optional("max_wal_senders"): int,
+                    Optional("max_worker_processes"): int
+                },
+                Optional("use_pg_rewind"): bool,
+                Optional("pg_hba"): [str],
+                Optional("pg_ident"): [str],
+                Optional("pg_ctl_timeout"): int,
+                Optional("use_slots"): bool,
+            },
+            Optional("primary_start_timeout"): int,
+            Optional("primary_stop_timeout"): int,
+            Optional("standby_cluster"): {
+                Or("host", "port", "restore_command"): Case({
+                    "host": str,
+                    "port": int,
+                    "restore_command": str
+                }),
+                Optional("primary_slot_name"): str,
+                Optional("create_replica_methods"): [str],
+                Optional("archive_cleanup_command"): str,
+                Optional("recovery_min_apply_delay"): str
+            },
+            Optional("synchronous_mode"): bool,
+            Optional("synchronous_mode_strict"): bool,
+            Optional("synchronous_node_count"): int
         },
-        Optional("initdb"): [Or(str, dict)]
+        Optional("initdb"): [Or(str, dict)],
+        Optional("method"): str
     },
     Or(*available_dcs): Case({
         "consul": {
             Or("host", "url"): Case({
                 "host": validate_host_port,
-                "url": str})
+                "url": str
+            }),
+            Optional("port"): int,
+            Optional("scheme"): str,
+            Optional("token"): str,
+            Optional("verify"): bool,
+            Optional("cacert"): str,
+            Optional("cert"): str,
+            Optional("key"): str,
+            Optional("dc"): str,
+            Optional("checks"): [str],
+            Optional("register_service"): bool,
+            Optional("service_tags"): [str],
+            Optional("service_check_interval"): str,
+            Optional("service_check_tls_server_name"): str,
+            Optional("consistency"): EnumValidator(('default', 'consistent', 'stale'),
+                                                   case_sensitive=True, raise_assert=True)
         },
         "etcd": validate_etcd,
         "etcd3": validate_etcd,
@@ -869,15 +981,28 @@ schema = Schema({
         },
         "zookeeper": {
             "hosts": Or(comma_separated_host_port, [validate_host_port]),
+            Optional("use_ssl"): bool,
+            Optional("cacert"): str,
+            Optional("cert"): str,
+            Optional("key"): str,
+            Optional("key_password"): str,
+            Optional("verify"): bool,
+            Optional("set_acls"): dict
         },
         "kubernetes": {
             "labels": {},
+            Optional("bypass_api_service"): bool,
             Optional("namespace"): str,
             Optional("scope_label"): str,
             Optional("role_label"): str,
+            Optional("leader_label_value"): str,
+            Optional("follower_label_value"): str,
+            Optional("standby_leader_label_value"): str,
+            Optional("tmp_role_label"): str,
             Optional("use_endpoints"): bool,
             Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address),
             Optional("ports"): [{"name": str, "port": int}],
+            Optional("cacert"): str,
             Optional("retriable_http_codes"): Or(int, [int]),
         },
     }),
@@ -915,7 +1040,8 @@ schema = Schema({
     },
     Optional("watchdog"): {
         Optional("mode"): validate_watchdog_mode,
-        Optional("device"): str
+        Optional("device"): str,
+        Optional("safety_margin"): int
     },
     Optional("tags"): {
         Optional("nofailover"): bool,

postgres0.yml

@@ -93,7 +93,7 @@ bootstrap:
   # Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter)
 # post_init: /usr/local/bin/setup_cluster.sh
 
-  # Some additional users users which needs to be created after initializing new cluster
+  # Some additional users which needs to be created after initializing new cluster
   users:
     admin:
       password: admin%

postgres1.yml

@@ -87,7 +87,7 @@ bootstrap:
   # Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter)
 # post_init: /usr/local/bin/setup_cluster.sh
 
-  # Some additional users users which needs to be created after initializing new cluster
+  # Some additional users which needs to be created after initializing new cluster
   users:
     admin:
       password: admin%

postgres2.yml

@@ -84,7 +84,7 @@ bootstrap:
   - encoding: UTF8
   - data-checksums
 
-  # Some additional users users which needs to be created after initializing new cluster
+  # Some additional users which needs to be created after initializing new cluster
   users:
     admin:
       password: admin%
@@ -121,4 +121,4 @@ tags:
     nofailover: false
     noloadbalance: false
     clonefrom: false
-    replicatefrom: postgres1
+#    replicatefrom: postgresql1

tests/test_api.py

@@ -226,7 +226,7 @@ class TestRestApiHandler(unittest.TestCase):
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary'))
         with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, None, None, '')])):
             self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
-        with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)),\
+        with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)), \
                 patch.object(GlobalConfig, 'is_paused', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /standby_leader')
 
@@ -559,7 +559,7 @@ class TestRestApiHandler(unittest.TestCase):
         request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2",' +\
                          ' "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
         MockRestApiServer(RestApiHandler, request)
-        with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)),\
+        with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)), \
                 patch.object(MockPatroni, 'dcs') as d:
             d.manual_failover.return_value = False
             MockRestApiServer(RestApiHandler, request)

tests/test_bootstrap.py

@@ -155,9 +155,9 @@ class TestBootstrap(BaseTestPostgresql):
 
         config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}}
 
-        with patch.object(Postgresql, 'is_running', Mock(return_value=False)),\
-                patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)),\
-                patch('multiprocessing.Process', Mock(side_effect=Exception)),\
+        with patch.object(Postgresql, 'is_running', Mock(return_value=False)), \
+                patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)), \
+                patch('multiprocessing.Process', Mock(side_effect=Exception)), \
                 patch('multiprocessing.get_context', Mock(side_effect=Exception), create=True):
             self.assertRaises(Exception, self.b.bootstrap, config)
         with open(os.path.join(self.p.data_dir, 'pg_hba.conf')) as f:
@@ -185,12 +185,12 @@ class TestBootstrap(BaseTestPostgresql):
         self.assertFalse(self.b.bootstrap(config))
 
         mock_cancellable_subprocess_call.return_value = 0
-        with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))),\
-                patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True),\
-                patch('os.path.isfile', Mock(return_value=True)),\
-                patch('os.unlink', Mock()),\
-                patch.object(ConfigHandler, 'save_configuration_files', Mock()),\
-                patch.object(ConfigHandler, 'restore_configuration_files', Mock()),\
+        with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))), \
+                patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True), \
+                patch('os.path.isfile', Mock(return_value=True)), \
+                patch('os.unlink', Mock()), \
+                patch.object(ConfigHandler, 'save_configuration_files', Mock()), \
+                patch.object(ConfigHandler, 'restore_configuration_files', Mock()), \
                 patch.object(ConfigHandler, 'write_recovery_conf', Mock()):
             with self.assertRaises(Exception) as e:
                 self.b.bootstrap(config)

tests/test_citus.py

@@ -56,7 +56,7 @@ class TestCitus(BaseTestPostgresql):
                                                'leader': 'leader', 'timeout': 30, 'cooldown': 10})
 
     def test_add_task(self):
-        with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
+        with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
                 patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)):
             self.c.add_task('', 1, self.cluster, '', None)
             mock_logger.assert_called_once()
@@ -120,7 +120,7 @@ class TestCitus(BaseTestPostgresql):
         self.c.process_tasks()
 
         self.c.add_task('after_promote', 0, self.cluster, self.cluster.leader_name, 'postgres://host3:5432/postgres')
-        with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
+        with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
                 patch.object(CitusHandler, 'query', Mock(side_effect=Exception)):
             self.c.process_tasks()
             mock_logger.assert_called_once()

tests/test_config.py

@@ -21,7 +21,8 @@ class TestConfig(unittest.TestCase):
         with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
             self.assertFalse(self.config.set_dynamic_configuration({'foo': 'bar'}))
         self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}, 'postgresql': {
-            'parameters': {'cluster_name': 1, 'wal_keep_size': 1, 'track_commit_timestamp': 1, 'wal_level': 1}}}))
+            'parameters': {'cluster_name': 1, 'hot_standby': 1, 'wal_keep_size': 1,
+                           'track_commit_timestamp': 1, 'wal_level': 1}}}))
 
     def test_reload_local_configuration(self):
         os.environ.update({

tests/test_ctl.py

@@ -22,7 +22,7 @@ from .test_ha import get_cluster_initialized_without_leader, get_cluster_initial
 
 
 @patch('patroni.ctl.load_config', Mock(return_value={
-    'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'},
+    'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'}, 'ctl': {'certfile': 'a'},
     'etcd': {'host': 'localhost:2379'}, 'citus': {'database': 'citus', 'group': 0},
     'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5}}))
 class TestCtl(unittest.TestCase):
@@ -83,9 +83,13 @@ class TestCtl(unittest.TestCase):
         scheduled_at = datetime.now(tzutc) + timedelta(seconds=600)
         cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at))
         del cluster.members[1].data['conn_url']
-        for fmt in ('pretty', 'json', 'yaml', 'tsv', 'topology'):
+        for fmt in ('pretty', 'json', 'yaml', 'topology'):
             self.assertIsNone(output_members({}, cluster, name='abc', fmt=fmt))
 
+        with patch('click.echo') as mock_echo:
+            self.assertIsNone(output_members({}, cluster, name='abc', fmt='tsv'))
+            self.assertEqual(mock_echo.call_args[0][0], 'abc\tother\t\tReplica\trunning\t\tunknown')
+
     @patch('patroni.ctl.get_dcs')
     @patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
     def test_switchover(self, mock_get_dcs):
@@ -451,7 +455,7 @@ class TestCtl(unittest.TestCase):
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
 
         for role in self.TEST_ROLES:
-            result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '-r', role], input='y')
+            result = self.runner.invoke(ctl, ['-k', 'flush', 'dummy', 'restart', '-r', role], input='y')
             assert 'No scheduled restart' in result.output
 
         result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force'])

tests/test_etcd.py

@@ -172,12 +172,12 @@ class TestClient(unittest.TestCase):
         self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
         self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
 
-        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
+        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \
                 patch.object(EtcdClient, '_load_machines_cache', Mock(side_effect=Exception)):
             self.client.http.request = Mock(side_effect=socket.error)
             self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
 
-        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
+        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \
                 patch.object(EtcdClient, '_load_machines_cache', Mock(return_value=True)):
             self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
 

tests/test_etcd3.py

@@ -5,8 +5,9 @@ import urllib3
 
 from mock import Mock, PropertyMock, patch
 from patroni.dcs.etcd import DnsCachingResolver
-from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\
-    InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3
+from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \
+    Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \
+    Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode
 from threading import Thread
 
 from . import SleepException, MockResponse
@@ -126,10 +127,16 @@ class TestPatroniEtcd3Client(BaseTestEtcd3):
         request = {'key': base64_encode('/patroni/test/leader')}
         mock_urlopen.return_value = MockResponse()
         mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}'
-        self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]})
         self.client.call_rpc('/kv/put', request)
         self.client.call_rpc('/kv/deleterange', request)
 
+    @patch.object(urllib3.PoolManager, 'urlopen')
+    def test_txn(self, mock_urlopen):
+        mock_urlopen.return_value = MockResponse()
+        mock_urlopen.return_value.content = '{"header":{"revision":"1"}}'
+        self.client.txn({'target': 'MOD', 'mod_revision': '1'},
+                        {'request_delete_range': {'key': base64_encode('/patroni/test/leader')}})
+
     @patch('time.time', Mock(side_effect=[1, 10.9, 100]))
     def test__wait_cache(self):
         with self.kv_cache.condition:
@@ -241,7 +248,7 @@ class TestEtcd3(BaseTestEtcd3):
             self.etcd3.update_leader(leader, '123', failsafe={'foo': 'bar'})
         self.etcd3._last_lease_refresh = 0
         self.etcd3.update_leader(leader, '124')
-        with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)),\
+        with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)), \
                 patch('time.time', Mock(side_effect=[0, 100, 200, 300])):
             self.assertRaises(Etcd3Error, self.etcd3.update_leader, leader, '126')
         self.etcd3._lease = leader.session

tests/test_ha.py

@@ -278,8 +278,10 @@ class TestHa(PostgresInit):
         self.p.follow = true
         self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
         self.p.is_running = true
+        ha_dcs_orig_name = self.ha.dcs.__class__.__name__
         self.ha.dcs.__class__.__name__ = 'Raft'
         self.assertEqual(self.ha.run_cycle(), 'started as a secondary')
+        self.ha.dcs.__class__.__name__ = ha_dcs_orig_name
 
     def test_recover_former_primary(self):
         self.p.follow = false
@@ -305,7 +307,7 @@ class TestHa(PostgresInit):
         self.p.is_running = false
         self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
         self.assertEqual(self.ha.run_cycle(), 'doing crash recovery in a single user mode')
-        with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)),\
+        with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)), \
                 patch.object(Ha, 'check_timeline', Mock(return_value=False)):
             self.ha._async_executor.schedule('doing crash recovery in a single user mode')
             self.ha.state_handler.cancellable._process = Mock()
@@ -338,7 +340,7 @@ class TestHa(PostgresInit):
         self.ha._rewind.check_leader_is_not_in_recovery = true
         with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)):
             self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader')
-        with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)),\
+        with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)), \
                 patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)):
             self.p.follow = true
             self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
@@ -373,6 +375,12 @@ class TestHa(PostgresInit):
     def test_acquire_lock_as_primary(self):
         self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
 
+    def test_leader_race_stale_primary(self):
+        with patch.object(Postgresql, 'get_primary_timeline', Mock(return_value=1)), \
+                patch('patroni.ha.logger.warning') as mock_logger:
+            self.assertEqual(self.ha.run_cycle(), 'demoting self because i am not the healthiest node')
+            self.assertEqual(mock_logger.call_args[0][0], 'My timeline %s is behind last known cluster timeline %s')
+
     def test_promoted_by_acquiring_lock(self):
         self.ha.is_healthiest_node = true
         self.p.is_leader = false
@@ -606,7 +614,7 @@ class TestHa(PostgresInit):
         self.e.initialize = true
         self.ha.bootstrap()
         self.p.is_leader = true
-        with patch.object(Watchdog, 'activate', Mock(return_value=False)),\
+        with patch.object(Watchdog, 'activate', Mock(return_value=False)), \
                 patch('patroni.ha.logger.error') as mock_logger:
             self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap')
             self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)
@@ -667,9 +675,9 @@ class TestHa(PostgresInit):
 
             self.ha.update_lock = false
             self.p.set_role('primary')
-            with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)),\
+            with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)), \
                     patch('patroni.async_executor.CriticalTask.result',
-                          PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True),\
+                          PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True), \
                     patch('patroni.postgresql.Postgresql.terminate_starting_postmaster') as mock_terminate:
                 self.assertEqual(self.ha.run_cycle(), 'lost leader lock during restart')
                 mock_terminate.assert_called()
@@ -883,7 +891,10 @@ class TestHa(PostgresInit):
         member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'})
         self.ha.fetch_node_status(member)
         member = Member(0, 'test', 1, {'api_url': 'http://localhost:8011/patroni'})
-        self.ha.fetch_node_status(member)
+        self.ha.patroni.request = Mock()
+        self.ha.patroni.request.return_value.data = b'{"wal":{"location":1},"role":"primary"}'
+        ret = self.ha.fetch_node_status(member)
+        self.assertFalse(ret.in_recovery)
 
     @patch.object(Rewind, 'pg_rewind', true)
     @patch.object(Rewind, 'check_leader_is_not_in_recovery', true)

tests/test_kubernetes.py

@@ -8,8 +8,8 @@ import unittest
 import urllib3
 
 from mock import Mock, PropertyMock, mock_open, patch
-from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\
-    K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\
+from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \
+    K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \
     Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
 from threading import Thread
 from . import MockResponse, SleepException
@@ -86,8 +86,8 @@ class TestK8sConfig(unittest.TestCase):
             with patch('os.environ', env):
                 self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config)
 
-        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
-                patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])),\
+        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \
+                patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])), \
                 patch('builtins.open', Mock(side_effect=[
                     mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')(),
                     mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')()])):
@@ -98,8 +98,8 @@ class TestK8sConfig(unittest.TestCase):
             self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a')
 
     def test_refresh_token(self):
-        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
-                patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])),\
+        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \
+                patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])), \
                 patch('builtins.open', Mock(side_effect=[
                     mock_open(read_data='cert')(), mock_open(read_data='a')(),
                     mock_open()(), mock_open(read_data='b')(), mock_open(read_data='c')()])):
@@ -138,10 +138,10 @@ class TestK8sConfig(unittest.TestCase):
 
         config["users"][0]["user"]["client-key-data"] = base64.b64encode(b'foobar').decode('utf-8')
         config["clusters"][0]["cluster"]["certificate-authority-data"] = base64.b64encode(b'foobar').decode('utf-8')
-        with patch('builtins.open', mock_open(read_data=json.dumps(config))),\
-                patch('os.write', Mock()), patch('os.close', Mock()),\
-                patch('os.remove') as mock_remove,\
-                patch('atexit.register') as mock_atexit,\
+        with patch('builtins.open', mock_open(read_data=json.dumps(config))), \
+                patch('os.write', Mock()), patch('os.close', Mock()), \
+                patch('os.remove') as mock_remove, \
+                patch('atexit.register') as mock_atexit, \
                 patch('tempfile.mkstemp') as mock_mkstemp:
             mock_mkstemp.side_effect = [(3, '1.tmp'), (4, '2.tmp')]
             k8s_config.load_kube_config()
@@ -298,6 +298,28 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         self.k.touch_member({'state': 'running', 'role': 'replica'})
         self.k.touch_member({'state': 'stopped', 'role': 'primary'})
 
+        self.k._role_label = 'isMaster'
+        self.k._leader_label_value = 'true'
+        self.k._follower_label_value = 'false'
+        self.k._standby_leader_label_value = 'false'
+        self.k._tmp_role_label = 'tmp_role'
+
+        self.k.touch_member({'state': 'running', 'role': 'replica'})
+        mock_patch_namespaced_pod.assert_called()
+        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'false')
+        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'replica')
+
+        self.k.touch_member({'state': 'running', 'role': 'standby-leader'})
+        mock_patch_namespaced_pod.assert_called()
+        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'false')
+        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'standby-leader')
+
+        self.k._name = 'p-0'
+        self.k.touch_member({'role': 'primary'})
+        mock_patch_namespaced_pod.assert_called()
+        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'true')
+        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'master')
+
     def test_initialize(self):
         self.k.initialize()
 

tests/test_log.py

@@ -43,7 +43,7 @@ class TestPatroniLogger(unittest.TestCase):
         _LOG.exception('test')
         logger.start()
 
-        with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)),\
+        with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)), \
                 patch('_pytest.logging.LogCaptureHandler.emit', Mock()):
             logging.error('test')
 

tests/test_postgresql.py

@@ -333,7 +333,7 @@ class TestPostgresql(BaseTestPostgresql):
 
         mock_read_auto = mock_open(read_data=read_data)
         mock_read_auto.return_value.__iter__ = lambda o: iter(o.readline, '')
-        with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])),\
+        with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])), \
                 patch('os.chmod', Mock()):
             self.p.config.write_postgresql_conf()
 
@@ -496,8 +496,8 @@ class TestPostgresql(BaseTestPostgresql):
             self.p.remove_data_directory()
         with patch('os.path.isfile', Mock(return_value=True)):
             self.p.remove_data_directory()
-        with patch('os.path.islink', Mock(side_effect=[False, False, True, True])),\
-                patch('os.listdir', Mock(return_value=['12345'])),\
+        with patch('os.path.islink', Mock(side_effect=[False, False, True, True])), \
+                patch('os.listdir', Mock(return_value=['12345'])), \
                 patch('os.path.realpath', Mock(side_effect=['../foo', '../foo_tsp'])):
             self.p.remove_data_directory()
 

tests/test_raft.py

@@ -4,7 +4,7 @@ import tempfile
 import time
 
 from mock import Mock, PropertyMock, patch
-from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL,\
+from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \
     Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
 from pysyncobj import SyncObjConf, FAIL_REASON
 

tests/test_rewind.py

@@ -65,14 +65,14 @@ class TestRewind(BaseTestPostgresql):
 
     def test_pg_rewind(self):
         r = {'user': '', 'host': '', 'port': '', 'database': '', 'password': ''}
-        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)),\
+        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)), \
                 patch.object(CancellableSubprocess, 'call', Mock(return_value=None)):
             with patch('subprocess.check_output', Mock(return_value=b'boo')):
                 self.assertFalse(self.r.pg_rewind(r))
             with patch('subprocess.check_output', Mock(side_effect=Exception)):
                 self.assertFalse(self.r.pg_rewind(r))
 
-        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)),\
+        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)), \
                 patch('subprocess.check_output', Mock(return_value=b'foo %f %p %r %% % %')):
             with patch.object(CancellableSubprocess, 'call', mock_cancellable_call):
                 self.assertFalse(self.r.pg_rewind(r))
@@ -91,7 +91,7 @@ class TestRewind(BaseTestPostgresql):
                                              'Latest checkpoint location': '0/'})):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
-        with patch.object(Postgresql, 'is_running', Mock(return_value=True)),\
+        with patch.object(Postgresql, 'is_running', Mock(return_value=True)), \
                 patch.object(MockCursor, 'fetchone',
                              Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
@@ -239,7 +239,7 @@ class TestRewind(BaseTestPostgresql):
 
         with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])):
             # successful archive_command call
-            with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)):
+            with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)) as mock_subprocess_call:
                 get_guc_value_res = [
                     'on', 'command %f',
                     'always', 'command %f',
@@ -252,6 +252,10 @@ class TestRewind(BaseTestPostgresql):
                                           '000000000000000000000000', 'command 000000000000000000000000'),
                                          mock_logger_info.call_args[0])
                         mock_logger_info.reset_mock()
+                        mock_subprocess_call.assert_called_once()
+                        self.assertEqual(mock_subprocess_call.call_args.args[0], ['command 000000000000000000000000'])
+                        self.assertEqual(mock_subprocess_call.call_args.kwargs['shell'], True)
+                        mock_subprocess_call.reset_mock()
 
             # failed archive_command call
             with patch.object(CancellableSubprocess, 'call', Mock(return_value=1)):

tests/test_slots.py

@@ -43,12 +43,12 @@ class TestSlotsHandler(BaseTestPostgresql):
         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')
-        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))),\
+        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
                 patch('patroni.postgresql.slots.logger.debug') as mock_debug:
             self.s.sync_replication_slots(cluster, False)
             mock_debug.assert_called_once()
         self.p.set_role('replica')
-        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)),\
+        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)), \
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
             self.s.sync_replication_slots(cluster, False, paused=True)
             mock_drop.assert_not_called()
@@ -93,11 +93,11 @@ class TestSlotsHandler(BaseTestPostgresql):
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')
         self.cluster.slots['ls'] = 12346
-        with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock()):
+        with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
             self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
-                patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))),\
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
+                patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
             self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
@@ -119,14 +119,14 @@ class TestSlotsHandler(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test_check_logical_slots_readiness(self):
         self.s.copy_logical_slots(self.cluster, ['ls'])
-        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
                 patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
-            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
-        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
                 patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))):
-            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
+            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))):
-            self.s.check_logical_slots_readiness(self.cluster, False, None)
+            self.s.check_logical_slots_readiness(self.cluster, None)
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
@@ -144,7 +144,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.assertRaises(OSError, fsync_dir, 'foo')
 
     def test_slots_advance_thread(self):
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
             self.s.schedule_advance_slots({'foo': {'bar': 100}})

tests/test_validator.py

@@ -15,7 +15,8 @@ config = {
     "scope": "string",
     "restapi": {
         "listen": "127.0.0.2:800",
-        "connect_address": "127.0.0.2:800"
+        "connect_address": "127.0.0.2:800",
+        "verify_client": 'none'
     },
     "bootstrap": {
         "dcs": {

tests/test_zookeeper.py

@@ -7,7 +7,7 @@ from kazoo.handlers.threading import SequentialThreadingHandler
 from kazoo.protocol.states import KeeperState, ZnodeStat
 from kazoo.retry import RetryFailedError
 from mock import Mock, PropertyMock, patch
-from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\
+from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient, \
     PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
 
 

typings/urllib3/__init__.pyi

@@ -1,6 +1,7 @@
+from .connectionpool import HTTPConnectionPool, HTTPSConnectionPool
 from .poolmanager import PoolManager
 from .response import HTTPResponse
 from .util.request import make_headers
 from .util.timeout import Timeout
 
-__all__ = ['HTTPResponse', 'PoolManager', 'Timeout', 'make_headers']
+__all__ = ['HTTPResponse', 'HTTPConnectionPool', 'HTTPSConnectionPool', 'PoolManager', 'Timeout', 'make_headers']

typings/urllib3/connectionpool.pyi

@@ -0,0 +1,2 @@
+class HTTPConnectionPool: ...
+class HTTPSConnectionPool(HTTPConnectionPool): ...