Release v3.3.0 (#3043)

* Make sure tests are not making external calls
and pass url with scheme to urllib3 to avoid warnings

* Make sure unit tests not rely on filesystem state

* Bump pyright and "solve" reported "issues"

Most of them are related to partially unknown types of values from empty
dict or list. To solve it for the empty dict we use `EMPTY_DICT` object of
newly introduced `_FrozenDict` class.

* Improve unit-tests code coverage

* Add release notes for 3.3.0

* Bump version

* Fix pyinstaller spec file

* python 3.6 compatibility

---------

Co-authored-by: Polina Bungina <[email protected]>

.github/workflows/tests.yaml

@@ -174,7 +174,7 @@ jobs:
 
     - uses: jakebailey/pyright-action@v1
       with:
-        version: 1.1.347
+        version: 1.1.356
 
   docs:
     runs-on: ubuntu-latest

Dockerfile

@@ -125,6 +125,8 @@ RUN if [ "$COMPRESS" = "true" ]; then \
         && /bin/busybox sh -c "(find $save_dirs -not -type d && cat /exclude /exclude && echo exclude) | sort | uniq -u | xargs /bin/busybox rm" \
         && /bin/busybox --install -s \
         && /bin/busybox sh -c "find $save_dirs -type d -depth -exec rmdir -p {} \; 2> /dev/null"; \
+    else \
+        /bin/busybox --install -s; \
     fi
 
 FROM scratch

docs/dynamic_configuration.rst

@@ -108,3 +108,7 @@ Note: if cluster topology is static (fixed number of nodes that never change the
 .. warning::
    Permanent replication slots are synchronized only from the ``primary``/``standby_leader`` to replica nodes. That means, applications are supposed to be using them only from the leader node. Using them on replica nodes will cause indefinite growth of ``pg_wal`` on all other nodes in the cluster.
    An exception to that rule are permanent physical slots that match the Patroni member names, if you happen to configure any. Those will be synchronized among all nodes as they are used for replication among them.
+
+
+.. warning::
+   Setting ``nostream`` tag on standby disables copying and synchronization of permanent logical replication slots on the node itself and all its cascading replicas if any.

docs/index.rst

@@ -28,12 +28,14 @@ Currently supported PostgreSQL versions: 9.3 to 16.
    patronictl
    replica_bootstrap
    replication_modes
+   standby_cluster
    watchdog
    pause
    dcs_failsafe_mode
    kubernetes
    citus
    existing_data
+   tools_integration
    security
    ha_multi_dc
    faq

docs/releases.rst

@@ -3,6 +3,76 @@
 Release notes
 =============
 
+Version 3.3.0
+-------------
+
+.. warning::
+   All older Partoni versions are not compatible with ``ydiff>=1.3``.
+
+   There are the following options available to "fix" the problem:
+
+   1. upgrade Patroni to the latest version
+   2. install ``ydiff<1.3`` after installing Patroni
+   3. install ``cdiff`` module
+
+
+**New features**
+
+- Add ability to pass ``auth_data`` to Zookeeper client (Aras Mumcuyan)
+
+  It allows to specify the authentication credentials to use for the connection.
+
+- Add a contrib script for ``Barman`` integration (Israel Barth Rubio)
+
+  Provide an application ``patroni_barman`` that allows to perform ``Barman`` operations remotely and can be used as a custom bootstrap/custom replica method or as an ``on_role_change`` callback. Please check :ref:`here <tools_integration>` for more information.
+
+- Support ``JSON`` log format (alisalemmi)
+
+  Apart from ``plain`` (default), Patroni now also supports ``json`` log format. Requires ``python-json-logger>=2.0.2`` library to be installed.
+
+- Show ``pending_restart_reason`` information (Polina Bungina)
+
+  Provide extended information about the PostgreSQL parameters that caused ``pending_restart`` flag to be set. Both ``patronictl list`` and ``/patroni`` REST API endpoint now show the parameters names and their "diff" as ``pending_restart_reason``.
+
+- Implement ``nostream`` tag (Grigory Smolkin)
+
+  If ``nostream`` tag is set to ``true``, the node will not use replication protocol to stream WAL but instead rely on archive recovery (if ``restore_command`` is configured). It also disables copying and synchronization of permanent logical replication slots on the node itself and all its cascading replicas.
+
+
+**Improvements**
+
+- Implement validation of the ``log`` section (Alexander Kukushkin)
+
+  Until now validator was not checking the correctness of the logging configuration provided.
+
+- Improve logging for PostgreSQL parameters change (Polina Bungina)
+
+  Convert old values to a human-readable format and log information about the ``pg_controldata`` vs Patroni global configuration mismatch.
+
+
+**Bugfixes**
+
+- Properly filter out not allowed ``pg_basebackup`` options (Israel Barth Rubio)
+
+  Due to a bug, Patroni was not properly filtering out the not allowed options configured for the ``basebackup`` replica bootstrap method, when provided in the ``- setting: value`` format.
+
+- Fix ``etcd3`` authentication error handling (Alexander Kukushkin)
+
+  Always retry one time on ``etcd3`` authentication error if authentication was not done right before executing the request. Also, do not restart watchers on reauthentication.
+
+- Improve logic of the validator files discovery (Waynerv)
+
+  Use ``importlib`` library to discover the files with available configuration parameters when possible (for Python 3.9+). This implementation is more stable and doesn't break the Patroni distributions based on ``zip`` archives.
+
+- Use ``target_session_attrs`` only when multiple hosts are specified in the ``standby_cluster`` section (Alexander Kukushkin)
+
+  ``target_session_attrs=read-write`` is now added to the ``primary_conninfo`` on the standby leader node only when ``standby_cluster.host`` section contains multiple hosts separated by commas.
+
+- Add compatibility code for ``ydiff`` library version 1.3+ (Alexander Kukushkin)
+
+  Patroni is relying on some API from ``ydiff`` that is not public because it is supposed to be just a terminal tool rather than a python module. Unfortunately, the API change in 1.3 broke old Patroni versions.
+
+
 Version 3.2.2
 -------------
 

docs/replica_bootstrap.rst

@@ -1,3 +1,5 @@
+.. _replica_imaging_and_bootstrap:
+
 Replica imaging and bootstrap
 =============================
 
@@ -79,14 +81,13 @@ As an example, you are able to bootstrap a fresh Patroni cluster from a Barman b
         method: barman
         barman:
             keep_existing_recovery_conf: true
-            command: patroni_barman_recover
-            api-url: https://barman-host:7480
+            command: patroni_barman --api-url https://barman-host:7480 recover
             barman-server: my_server
             ssh-command: ssh postgres@patroni-host
 
 .. note::
-    ``patroni_barman_recover`` requires that you have both Barman and ``pg-backup-api`` configured in the Barman host, so it can execute a remote ``barman recover`` through the backup API.
-    The above example uses a subset of the available parameters. You can get more information running ``patroni_barman_recover --help``.
+    ``patroni_barman recover`` requires that you have both Barman and ``pg-backup-api`` configured in the Barman host, so it can execute a remote ``barman recover`` through the backup API.
+    The above example uses a subset of the available parameters. You can get more information running ``patroni_barman recover --help``.
 
 .. _custom_replica_creation:
 
@@ -150,16 +151,15 @@ example: Barman
             - barman
             - basebackup
         barman:
-            command: patroni_barman_recover
-            api-url: https://barman-host:7480
+            command: patroni_barman --api-url https://barman-host:7480 recover
             barman-server: my_server
             ssh-command: ssh postgres@patroni-host
         basebackup:
             max-rate: '100M'
 
 .. note::
-    ``patroni_barman_recover`` requires that you have both Barman and ``pg-backup-api`` configured in the Barman host, so it can execute a remote ``barman recover`` through the backup API.
-    The above example uses a subset of the available parameters. You can get more information running ``patroni_barman_recover --help``.
+    ``patroni_barman recover`` requires that you have both Barman and ``pg-backup-api`` configured in the Barman host, so it can execute a remote ``barman recover`` through the backup API.
+    The above example uses a subset of the available parameters. You can get more information running ``patroni_barman recover --help``.
 
 The ``create_replica_methods`` defines available replica creation methods and the order of executing them. Patroni will
 stop on the first one that returns 0. Each method should define a separate section in the configuration file, listing the command
@@ -219,59 +219,3 @@ and
             - waldir: /pg-wal-mount/external-waldir
 
 If all replica creation methods fail, Patroni will try again all methods in order during the next event loop cycle.
-
-.. _standby_cluster:
-
-Standby cluster
----------------
-
-Another available option is to run a "standby cluster", that contains only of
-standby nodes replicating from some remote node. This type of clusters has:
-
-* "standby leader", that behaves pretty much like a regular cluster leader,
-  except it replicates from a remote node.
-
-* cascade replicas, that are replicating from standby leader.
-
-Standby leader holds and updates a leader lock in DCS. If the leader lock
-expires, cascade replicas will perform an election to choose another leader
-from the standbys.
-
-There is no further relationship between the standby cluster and the primary
-cluster it replicates from, in particular, they must not share the same DCS
-scope if they use the same DCS. They do not know anything else from each other
-apart from replication information. Also, the standby cluster is not being
-displayed in :ref:`patronictl_list` or :ref:`patronictl_topology` output on the
-primary cluster.
-
-For the sake of flexibility, you can specify methods of creating a replica and
-recovery WAL records when a cluster is in the "standby mode" by providing
-`create_replica_methods` key in `standby_cluster` section. It is distinct from
-creating replicas, when cluster is detached and functions as a normal cluster,
-which is controlled by `create_replica_methods` in `postgresql` section. Both
-"standby" and "normal" `create_replica_methods` reference  keys in `postgresql`
-section.
-
-To configure such cluster you need to specify the section ``standby_cluster``
-in a patroni configuration:
-
-.. code:: YAML
-
-    bootstrap:
-        dcs:
-            standby_cluster:
-                host: 1.2.3.4
-                port: 5432
-                primary_slot_name: patroni
-                create_replica_methods:
-                - basebackup
-
-Note, that these options will be applied only once during cluster bootstrap,
-and the only way to change them afterwards is through DCS.
-
-Patroni expects to find `postgresql.conf` or `postgresql.conf.backup` in PGDATA
-of the remote primary and will not start if it does not find it after a
-basebackup. If the remote primary keeps its `postgresql.conf` elsewhere, it is
-your responsibility to copy it to PGDATA.
-
-If you use replication slots on the standby cluster, you must also create the corresponding replication slot on the primary cluster.  It will not be done automatically by the standby cluster implementation.  You can use Patroni's permanent replication slots feature on the primary cluster to maintain a replication slot with the same name as ``primary_slot_name``, or its default value if ``primary_slot_name`` is not provided.

docs/replication_modes.rst

@@ -53,7 +53,7 @@ are available. As a downside, the primary is not be available for writes
 blocking all client write requests until at least one synchronous replica comes
 up.
 
-You can ensure that a standby never becomes the synchronous standby by setting ``nosync`` tag to true. This is recommended to set for standbys that are behind slow network connections and would cause performance degradation when becoming a synchronous standby.
+You can ensure that a standby never becomes the synchronous standby by setting ``nosync`` tag to true. This is recommended to set for standbys that are behind slow network connections and would cause performance degradation when becoming a synchronous standby. Setting tag ``nostream`` to true will also have the same effect.
 
 Synchronous mode can be switched on and off via Patroni REST interface. See :ref:`dynamic configuration <dynamic_configuration>` for instructions.
 

docs/standby_cluster.rst

@@ -0,0 +1,78 @@
+.. _standby_cluster:
+
+Standby cluster
+---------------
+
+Patroni also support running cascading replication to a remote datacenter
+(region) using a feature that is called "standby cluster". This type of
+clusters has:
+
+* "standby leader", that behaves pretty much like a regular cluster leader,
+  except it replicates from a remote node.
+
+* cascade replicas, that are replicating from standby leader.
+
+Standby leader holds and updates a leader lock in DCS. If the leader lock
+expires, cascade replicas will perform an election to choose another leader
+from the standbys.
+
+There is no further relationship between the standby cluster and the primary
+cluster it replicates from, in particular, they must not share the same DCS
+scope if they use the same DCS. They do not know anything else from each other
+apart from replication information. Also, the standby cluster is not being
+displayed in :ref:`patronictl_list` or :ref:`patronictl_topology` output on the
+primary cluster.
+
+For the sake of flexibility, you can specify methods of creating a replica and
+recovery WAL records when a cluster is in the "standby mode" by providing
+:ref:`create_replica_methods <custom_replica_creation>` key in
+`standby_cluster` section. It is distinct from creating replicas, when cluster
+is detached and functions as a normal cluster, which is controlled by
+`create_replica_methods` in `postgresql` section. Both "standby" and "normal"
+`create_replica_methods` reference  keys in `postgresql` section.
+
+To configure such cluster you need to specify the section ``standby_cluster``
+in a patroni configuration:
+
+.. code:: YAML
+
+    bootstrap:
+        dcs:
+            standby_cluster:
+                host: 1.2.3.4
+                port: 5432
+                primary_slot_name: patroni
+                create_replica_methods:
+                - basebackup
+
+Note, that these options will be applied only once during cluster bootstrap,
+and the only way to change them afterwards is through DCS.
+
+Patroni expects to find `postgresql.conf` or `postgresql.conf.backup` in PGDATA
+of the remote primary and will not start if it does not find it after a
+basebackup. If the remote primary keeps its `postgresql.conf` elsewhere, it is
+your responsibility to copy it to PGDATA.
+
+If you use replication slots on the standby cluster, you must also create the
+corresponding replication slot on the primary cluster.  It will not be done
+automatically by the standby cluster implementation.  You can use Patroni's
+permanent replication slots feature on the primary cluster to maintain a
+replication slot with the same name as ``primary_slot_name``, or its default
+value if ``primary_slot_name`` is not provided.
+
+In case the remote site doesn't provide a single endpoint that connects to a
+primary, one could list all hosts of the source cluster in the
+``standby_cluster.host`` section.  When ``standby_cluster.host`` contains
+multiple hosts separated by commas, Patroni will:
+
+* add ``target_session_attrs=read-write`` to the ``primary_conninfo`` on the
+  standby leader node.
+* use ``target_session_attrs=read-write`` when trying to determine whether we
+  need to run ``pg_rewind`` or when executing ``pg_rewind`` on all nodes of the
+  standby cluster.
+  
+There is also a possibility to replicate the standby cluster from another
+standby cluster or from a standby member of the primary cluster: for that, you
+need to define a single host in the ``standby_cluster.host`` section. However,
+you need to beware that in this case ``pg_rewind`` will fail to execute on the
+standby cluster.

docs/tools_integration.rst

@@ -0,0 +1,64 @@
+.. _tools_integration:
+
+Integration with other tools
+============================
+
+Patroni is able to integrate with other tools in your stack. In this section you
+will find a list of examples, which although not an exhaustive list, might
+provide you with ideas on how Patroni can integrate with other tools.
+
+Barman
+------
+
+Patroni delivers an application named ``patroni_barman`` which has logic to
+communicate with ``pg-backup-api``, so you are able to perform Barman operations
+remotely.
+
+This application currently has a couple of sub-commands: ``recover`` and
+``config-switch``.
+
+patroni_barman recover
+^^^^^^^^^^^^^^^^^^^^^^
+
+The ``recover`` sub-command can be used as a custom bootstrap or custom replica
+creation method. You can find more information about that in
+:ref:`replica_imaging_and_bootstrap`.
+
+patroni_barman config-switch
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The ``config-switch`` sub-command is designed to be used as an ``on_role_change``
+callback in Patroni. As an example, assume you are streaming WALs from your
+current primary to your Barman host. In the event of a failover in the cluster
+you might want to start streaming WALs from the new primary. You can accomplish
+this by using ``patroni_barman config-switch`` as the ``on_role_change`` callback.
+
+.. note::
+    That sub-command relies on the ``barman config-switch`` command, which is in
+    charge of overriding the configuration of a Barman server by applying a
+    pre-defined model on top of it. This command is available since Barman 3.10.
+    Please consult the Barman documentation for more details.
+
+This is an example of how you can configure Patroni to apply a configuration
+model in case this Patroni node is promoted to primary:
+
+.. code:: YAML
+
+    postgresql:
+        callbacks:
+            on_role_change: >
+                patroni_barman
+                    --api-url YOUR_API_URL
+                    config-switch
+                    --barman-server YOUR_BARMAN_SERVER_NAME
+                    --barman-model YOUR_BARMAN_MODEL_NAME
+                    --switch-when promoted
+
+.. note::
+    ``patroni_barman config-switch`` requires that you have both Barman and
+    ``pg-backup-api`` configured in the Barman host, so it can execute a remote
+    ``barman config-switch`` through the backup API. Also, it requires that you
+    have pre-configured Barman models to be applied. The above example uses a
+    subset of the available parameters. You can get more information running
+    ``patroni_barman config-switch --help``, and by consulting the Barman
+    documentation.

docs/yaml_configuration.rst

@@ -399,6 +399,7 @@ Tags
 -  **nosync**: ``true`` or ``false``. If set to ``true`` the node will never be selected as a synchronous replica.
 -  **nofailover**: ``true`` or ``false``, controls whether this node is allowed to participate in the leader race and become a leader. Defaults to ``false``, meaning this node _can_ participate in leader races. 
 -  **failover_priority**: integer, controls the priority that this node should have during failover. Nodes with higher priority will be preferred over lower priority nodes if they received/replayed the same amount of WAL. However, nodes with higher values of receive/replay LSN are preferred regardless of their priority. If the ``failover_priority`` is 0 or negative - such node is not allowed to participate in the leader race and to become a leader (similar to ``nofailover: true``).
+-  **nostream**: ``true`` or ``false``. If set to ``true`` the node will not use replication protocol to stream WAL. It will rely instead on archive recovery (if ``restore_command`` is configured) and ``pg_wal``/``pg_xlog`` polling. It also disables copying and synchronization of permanent logical replication slots on the node itself and all its cascading replicas. Setting this tag on primary node has no effect.
 
 .. warning::
    Provide only one of ``nofailover`` or ``failover_priority``. Providing ``nofailover: true`` is the same as ``failover_priority: 0``, and providing ``nofailover: false`` will give the node priority 1. 

features/environment.py

@@ -256,10 +256,18 @@ class PatroniController(AbstractController):
                         'parameters': {
                             'wal_keep_segments': 100,
                             'archive_mode': 'on',
-                            'archive_command': (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT
-                                                + ' --mode archive '
-                                                + '--dirname {} --filename %f --pathname %p').format(
-                                                    os.path.join(self._work_directory, 'data', 'wal_archive'))
+                            'archive_command':
+                                (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT
+                                 + ' --mode archive '
+                                 + '--dirname {} --filename %f --pathname %p').format(
+                                     os.path.join(self._work_directory, 'data',
+                                                  f'wal_archive{str(self._citus_group or "")}')).replace('\\', '/'),
+                            'restore_command':
+                                (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT
+                                 + ' --mode restore '
+                                 + '--dirname {} --filename %f --pathname %p').format(
+                                     os.path.join(self._work_directory, 'data',
+                                                  f'wal_archive{str(self._citus_group or "")}')).replace('\\', '/')
                         }
                     }
                 }
@@ -928,11 +936,6 @@ class PatroniPoolController(object):
         custom_config = {
             'scope': cluster_name,
             'postgresql': {
-                'recovery_conf': {
-                    'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore '
-                                        + '--dirname {} --filename %f --pathname %p')
-                    .format(os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/'))
-                },
                 'create_replica_methods': ['no_leader_bootstrap'],
                 'no_leader_bootstrap': self.backup_restore_config({'no_leader': '1'})
             }

features/nostream_node.feature

@@ -0,0 +1,18 @@
+Feature: nostream node
+
+Scenario: check nostream node is recovering from archive
+    When I start postgres0
+    And I configure and start postgres1 with a tag nostream true
+    Then "members/postgres1" key in DCS has replication_state=in archive recovery after 10 seconds
+    And replication works from postgres0 to postgres1 after 30 seconds
+
+@slot-advance
+Scenario: check permanent logical replication slots are not copied
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}}, "slots":{"test_logical":{"type":"logical","database":"postgres","plugin":"test_decoding"}}}
+    Then I receive a response code 200
+    When I run patronictl.py restart batman postgres0 --force
+    Then postgres0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
+    When I configure and start postgres2 with a tag replicatefrom postgres1
+    Then "members/postgres2" key in DCS has replication_state=streaming after 10 seconds 
+    And postgres1 does not have a replication slot named test_logical
+    And postgres2 does not have a replication slot named test_logical

features/standby_cluster.feature

@@ -26,7 +26,7 @@ Feature: standby cluster
   Scenario: Detach exiting node from the cluster
     When I shut down postgres1
     Then postgres0 is a leader after 10 seconds
-    And "members/postgres0" key in DCS has role=master after 3 seconds
+    And "members/postgres0" key in DCS has role=master after 5 seconds
     When I issue a GET request to http://127.0.0.1:8008/
     Then I receive a response code 200
 
@@ -47,6 +47,7 @@ Feature: standby cluster
     And there is a postgres1_cb.log with "on_role_change standby_leader batman1" in postgres1 data directory
     When I start postgres2 in a cluster batman1
     Then postgres2 role is the replica after 24 seconds
+    And postgres2 is replicating from postgres1 after 10 seconds
     And table foo is present on postgres2 after 20 seconds
     When I issue a GET request to http://127.0.0.1:8010/patroni
     Then I receive a response code 200

features/steps/basic_replication.py

@@ -46,11 +46,17 @@ def kill_postgres(context, name):
     return context.pctl.stop(name, kill=True, postgres=True)
 
 
+def get_wal_name(context, pg_name):
+    version = context.pctl.query(pg_name, "SHOW server_version_num").fetchone()[0]
+    return 'xlog' if int(version) / 10000 < 10 else 'wal'
+
+
 @step('I add the table {table_name:w} to {pg_name:w}')
 def add_table(context, table_name, pg_name):
     # parse the configuration file and get the port
     try:
         context.pctl.query(pg_name, "CREATE TABLE public.{0}()".format(table_name))
+        context.pctl.query(pg_name, "SELECT pg_switch_{0}()".format(get_wal_name(context, pg_name)))
     except pg.Error as e:
         assert False, "Error creating table {0} on {1}: {2}".format(table_name, pg_name, e)
 
@@ -59,9 +65,7 @@ def add_table(context, table_name, pg_name):
 def toggle_wal_replay(context, action, pg_name):
     # pause or resume the wal replay process
     try:
-        version = context.pctl.query(pg_name, "SHOW server_version_num").fetchone()[0]
-        wal_name = 'xlog' if int(version) / 10000 < 10 else 'wal'
-        context.pctl.query(pg_name, "SELECT pg_{0}_replay_{1}()".format(wal_name, action))
+        context.pctl.query(pg_name, "SELECT pg_{0}_replay_{1}()".format(get_wal_name(context, pg_name), action))
     except pg.Error as e:
         assert False, "Error during {0} wal recovery on {1}: {2}".format(action, pg_name, e)
 

patroni.spec

@@ -13,13 +13,17 @@ def hiddenimports():
         sys.path.pop(0)
 
 
+def resources():
+    import os
+    res_dir = 'patroni/postgresql/available_parameters/'
+    exts = set(f.split('.')[-1] for f in os.listdir(res_dir))
+    return [(res_dir + '*.' + e, res_dir) for e in exts if e.lower() in {'yml', 'yaml'}]
+
+
 a = Analysis(['patroni/__main__.py'],
              pathex=[],
              binaries=None,
-             datas=[
-                ('patroni/postgresql/available_parameters/*.yml', 'patroni/postgresql/available_parameters'),
-                ('patroni/postgresql/available_parameters/*.yaml', 'patroni/postgresql/available_parameters'),
-            ],
+             datas=resources(),
              hiddenimports=hiddenimports(),
              hookspath=[],
              runtime_hooks=[],

patroni/collections.py

@@ -1,9 +1,10 @@
 """Patroni custom object types somewhat like :mod:`collections` module.
 
-Provides a case insensitive :class:`dict` and :class:`set` object types.
+Provides a case insensitive :class:`dict` and :class:`set` object types, and `EMPTY_DICT` frozen dictionary object.
 """
 from collections import OrderedDict
-from typing import Any, Collection, Dict, Iterator, KeysView, MutableMapping, MutableSet, Optional
+from copy import deepcopy
+from typing import Any, Collection, Dict, Iterator, KeysView, Mapping, MutableMapping, MutableSet, Optional
 
 
 class CaseInsensitiveSet(MutableSet[str]):
@@ -48,7 +49,7 @@ class CaseInsensitiveSet(MutableSet[str]):
         """
         return str(set(self._values.values()))
 
-    def __contains__(self, value: str) -> bool:
+    def __contains__(self, value: object) -> bool:
         """Check if set contains *value*.
 
         The check is performed case-insensitively.
@@ -57,7 +58,7 @@ class CaseInsensitiveSet(MutableSet[str]):
 
         :returns: ``True`` if *value* is already in the set, ``False`` otherwise.
         """
-        return value.lower() in self._values
+        return isinstance(value, str) and value.lower() in self._values
 
     def __iter__(self) -> Iterator[str]:
         """Iterate over the values in this set.
@@ -207,3 +208,47 @@ class CaseInsensitiveDict(MutableMapping[str, Any]):
             "<CaseInsensitiveDict{'A': 'B', 'c': 'd'} at ..."
         """
         return '<{0}{1} at {2:x}>'.format(type(self).__name__, dict(self.items()), id(self))
+
+
+class _FrozenDict(Mapping[str, Any]):
+    """Frozen dictionary object."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Create a new instance of :class:`_FrozenDict` with given data."""
+        self.__values: Dict[str, Any] = dict(*args, **kwargs)
+
+    def __iter__(self) -> Iterator[str]:
+        """Iterate over keys of this dict.
+
+        :yields: each key present in the dict. Yields each key with its last case that has been stored.
+        """
+        return iter(self.__values)
+
+    def __len__(self) -> int:
+        """Get the length of this dict.
+
+        :returns: number of keys in the dict.
+
+        :Example:
+
+        >>> len(_FrozenDict())
+        0
+        """
+        return len(self.__values)
+
+    def __getitem__(self, key: str) -> Any:
+        """Get the value corresponding to *key*.
+
+        :returns: value corresponding to *key*.
+        """
+        return self.__values[key]
+
+    def copy(self) -> Dict[str, Any]:
+        """Create a copy of this dict.
+
+        :return: a new dict object with the same keys and values of this dict.
+        """
+        return deepcopy(self.__values)
+
+
+EMPTY_DICT = _FrozenDict()

patroni/config.py

@@ -12,7 +12,7 @@ from copy import deepcopy
 from typing import Any, Callable, Collection, Dict, List, Optional, Union, TYPE_CHECKING
 
 from . import PATRONI_ENV_PREFIX
-from .collections import CaseInsensitiveDict
+from .collections import CaseInsensitiveDict, EMPTY_DICT
 from .dcs import ClusterConfig
 from .exceptions import ConfigParseError
 from .file_perm import pg_perm
@@ -445,14 +445,14 @@ class Config(object):
 
         for name, value in dynamic_configuration.items():
             if name == 'postgresql':
-                for name, value in (value or {}).items():
+                for name, value in (value or EMPTY_DICT).items():
                     if name == 'parameters':
                         config['postgresql'][name].update(self._process_postgresql_parameters(value))
                     elif name not in ('connect_address', 'proxy_address', 'listen',
                                       'config_dir', 'data_dir', 'pgpass', 'authentication'):
                         config['postgresql'][name] = deepcopy(value)
             elif name == 'standby_cluster':
-                for name, value in (value or {}).items():
+                for name, value in (value or EMPTY_DICT).items():
                     if name in self.__DEFAULT_CONFIG['standby_cluster']:
                         config['standby_cluster'][name] = deepcopy(value)
             elif name in config:  # only variables present in __DEFAULT_CONFIG allowed to be overridden from DCS

patroni/config_generator.py

@@ -15,6 +15,7 @@ if TYPE_CHECKING:  # pragma: no cover
     from psycopg2 import cursor
 
 from . import psycopg
+from .collections import EMPTY_DICT
 from .config import Config
 from .exceptions import PatroniException
 from .log import PatroniLogger
@@ -126,6 +127,7 @@ class AbstractConfigGenerator(abc.ABC):
                 'noloadbalance': False,
                 'clonefrom': True,
                 'nosync': False,
+                'nostream': False,
             }
         }
 
@@ -243,7 +245,8 @@ class SampleConfigGenerator(AbstractConfigGenerator):
                   See :func:`~patroni.postgresql.misc.postgres_major_version_to_int` and
                   :func:`~patroni.utils.get_major_version`.
         """
-        postgres_bin = ((self.config.get('postgresql') or {}).get('bin_name') or {}).get('postgres', 'postgres')
+        postgres_bin = ((self.config.get('postgresql')
+                         or EMPTY_DICT).get('bin_name') or EMPTY_DICT).get('postgres', 'postgres')
         return postgres_major_version_to_int(get_major_version(self.config['postgresql'].get('bin_dir'), postgres_bin))
 
     def generate(self) -> None:
@@ -410,8 +413,10 @@ class RunningClusterConfigGenerator(AbstractConfigGenerator):
             val = self.parsed_dsn.get(conn_param, os.getenv(env_var))
             if val:
                 su_params[conn_param] = val
-        patroni_env_su_username = ((self.config.get('authentication') or {}).get('superuser') or {}).get('username')
-        patroni_env_su_pwd = ((self.config.get('authentication') or {}).get('superuser') or {}).get('password')
+        patroni_env_su_username = ((self.config.get('authentication')
+                                    or EMPTY_DICT).get('superuser') or EMPTY_DICT).get('username')
+        patroni_env_su_pwd = ((self.config.get('authentication')
+                               or EMPTY_DICT).get('superuser') or EMPTY_DICT).get('password')
         # because we use "username" in the config for some reason
         su_params['username'] = su_params.pop('user', patroni_env_su_username) or getuser()
         su_params['password'] = su_params.get('password', patroni_env_su_pwd) or \

patroni/ctl.py

@@ -41,8 +41,12 @@ if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
     from psycopg2 import cursor
 
-try:
-    from ydiff import markup_to_pager, PatchStream  # pyright: ignore [reportMissingModuleSource]
+try:  # pragma: no cover
+    from ydiff import markup_to_pager  # pyright: ignore [reportMissingModuleSource]
+    try:
+        from ydiff import PatchStream  # pyright: ignore [reportMissingModuleSource]
+    except ImportError:
+        PatchStream = iter
 except ImportError:  # pragma: no cover
     from cdiff import markup_to_pager, PatchStream  # pyright: ignore [reportMissingModuleSource]
 

patroni/dcs/__init__.py

@@ -85,6 +85,8 @@ def dcs_modules() -> List[str]:
 
     :returns: list of known module names with absolute python module path namespace, e.g. ``patroni.dcs.etcd``.
     """
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(__package__, str)
     return iter_modules(__package__)
 
 
@@ -101,6 +103,8 @@ def iter_dcs_classes(
 
     :returns: an iterator of tuples, each containing the module ``name`` and the imported DCS class object.
     """
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(__package__, str)
     return iter_classes(__package__, AbstractDCS, config)
 
 
@@ -1039,6 +1043,8 @@ class Cluster(NamedTuple('Cluster',
         .. note::
             Permanent replication slots are only considered if ``use_slots`` configuration is enabled.
             A node that is not supposed to become a leader (*nofailover*) will not have permanent replication slots.
+            Also node with disabled streaming (*nostream*) and its cascading followers must not have permanent
+            logical slots due to lack of feedback from node to primary, which makes them unsafe to use.
 
             In a standby cluster we only support physical replication slots.
 
@@ -1054,7 +1060,7 @@ class Cluster(NamedTuple('Cluster',
         if not global_config.use_slots or tags.nofailover:
             return {}
 
-        if global_config.is_standby_cluster:
+        if global_config.is_standby_cluster or self.get_slot_name_on_primary(postgresql.name, tags) is None:
             return self.__permanent_physical_slots \
                 if postgresql.major_version >= SLOT_ADVANCE_AVAILABLE_VERSION or role == 'standby_leader' else {}
 
@@ -1069,6 +1075,10 @@ class Cluster(NamedTuple('Cluster',
         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.
 
+        If the ``nostream`` tag is set on the member - we should not create the replication slot for it on
+        the current primary or any other member even if ``replicatefrom`` is set, because ``nostream`` disables
+        WAL streaming.
+
         Will log an error if:
 
             * Conflicting slot names between members are found
@@ -1083,8 +1093,9 @@ class Cluster(NamedTuple('Cluster',
         if not global_config.use_slots:
             return {}
 
-        # we always want to exclude the member with our name from the list
-        members = filter(lambda m: m.name != name, self.members)
+        # we always want to exclude the member with our name from the list,
+        # also exlude members with disabled WAL streaming
+        members = filter(lambda m: m.name != name and not m.nostream, self.members)
 
         if role in ('master', 'primary', 'standby_leader'):
             members = [m for m in members if m.replicatefrom is None
@@ -1172,7 +1183,7 @@ class Cluster(NamedTuple('Cluster',
             return any(self.should_enforce_hot_standby_feedback(postgresql, m) for m in members)
         return False
 
-    def get_slot_name_on_primary(self, name: str, tags: Tags) -> str:
+    def get_slot_name_on_primary(self, name: str, tags: Tags) -> Optional[str]:
         """Get the name of physical replication slot for this node on the primary.
 
         .. note::
@@ -1186,6 +1197,8 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: the slot name on the primary that is in use for physical replication on this node.
         """
+        if tags.nostream:
+            return None
         replicatefrom = self.get_member(tags.replicatefrom, False) if tags.replicatefrom else None
         return self.get_slot_name_on_primary(replicatefrom.name, replicatefrom) \
             if isinstance(replicatefrom, Member) else slot_name_from_member_name(name)

patroni/dcs/consul.py

@@ -42,57 +42,6 @@ class InvalidSession(ConsulException):
     """invalid session"""
 
 
-class ConsulAgentService(base.Consul.Agent.Service):
-    """
-    Consul.Agent.Session with support of ``tagged_addresses``.
-
-    We do it in the Patroni code because ``python-consul`` and
-    ``python-consul2`` modules don't receive any updates for at least 3 years.
-    """
-
-    def register(self, name: str, service_id: Optional[str] = None, address: Optional[str] = None,
-                 port: Optional[int] = None, tags: Optional[List[str]] = None, check: Optional[Dict[str, str]] = None,
-                 token: Optional[str] = None, enable_tag_override: bool = False,
-                 tagged_addresses: Optional[Dict[str, Dict[str, Union[str, int]]]] = None, **kwargs: Any) -> bool:
-        """Add a new service to the local agent.
-
-        :param name: name of the service.
-        :param service_id: service id, optional, if not provided *name* is used.
-        :param address: will default to the address of the agent if not provided.
-        :param port: port on which the service is available.
-        :param tagged_addresses: additional addresses for a node or service.
-        :tags: a list of string values that add service-level labels.
-        :enable_tag_override: optional ``bool`` that enable you to modify a service tags from servers
-                              (consul agent role server). Default is set to ``False``.
-        :check: an optional health check for this service.
-        :token: an optional ACL token to apply to this request.
-
-        :returns: ``True`` if the service was successfully registered/updated, otherwise ``False``.
-        """
-        payload: Dict[str, Any] = {'name': name}
-
-        if enable_tag_override:
-            payload['enabletagoverride'] = enable_tag_override
-        if service_id:
-            payload['id'] = service_id
-        if address:
-            payload['address'] = address
-        if port:
-            payload['port'] = port
-        if tagged_addresses:
-            payload['tagged_addresses'] = tagged_addresses
-        if tags:
-            payload['tags'] = tags
-        if check:
-            payload['check'] = check
-
-        token = token or self.agent.token
-        params = {'token': token} if token else {}
-
-        return self.agent.http.put(base.CB.bool(), '/v1/agent/service/register',
-                                   params=params, data=json.dumps(payload))
-
-
 class Response(NamedTuple):
     code: int
     headers: Union[Mapping[str, str], Mapping[bytes, bytes], None]
@@ -320,7 +269,6 @@ class Consul(AbstractDCS):
             kwargs['verify'] = verify
 
         self._client = ConsulClient(**kwargs)
-        self._agent_service = ConsulAgentService(self._client)
         self.set_retry_timeout(config['retry_timeout'])
         self.set_ttl(config.get('ttl') or 30)
         self._last_session_refresh = 0
@@ -496,8 +444,9 @@ class Consul(AbstractDCS):
 
         :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values.
         """
+        results: Optional[List[Dict[str, Any]]]
         _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency)
-        clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict)
+        clusters: Dict[int, Dict[str, Dict[str, Any]]] = defaultdict(dict)
         for node in results or []:
             key = node['Key'][len(path):].split('/', 1)
             if len(key) == 2 and self._mpp.group_re.match(key[0]):
@@ -555,14 +504,14 @@ class Consul(AbstractDCS):
     @catch_consul_errors
     def register_service(self, service_name: str, **kwargs: Any) -> bool:
         logger.info('Register service %s, params %s', service_name, kwargs)
-        return self._agent_service.register(service_name, **kwargs)
+        return self._client.agent.service.register(service_name, **kwargs)
 
     @catch_consul_errors
     def deregister_service(self, service_id: str) -> bool:
         logger.info('Deregister service %s', service_id)
         # service_id can contain special characters, but is used as part of uri in deregister request
         service_id = quote(service_id)
-        return self._agent_service.deregister(service_id)
+        return self._client.agent.service.deregister(service_id)
 
     def _update_service(self, data: Dict[str, Any]) -> Optional[bool]:
         service_name = self._service_name
@@ -629,14 +578,17 @@ class Consul(AbstractDCS):
         try:
             return retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session)
         except InvalidSession:
-            logger.error('Our session disappeared from Consul. Will try to get a new one and retry attempt')
             self._session = None
-            retry.ensure_deadline(0)
+
+            if not retry.ensure_deadline(0):
+                logger.error('Our session disappeared from Consul. Deadline exceeded, giving up')
+                return False
+
+            logger.error('Our session disappeared from Consul. Will try to get a new one and retry attempt')
 
             retry(self._do_refresh_session)
 
             retry.ensure_deadline(1, ConsulError('_do_attempt_to_acquire_leader timeout'))
-
             return retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session)
 
     @catch_return_false_exception

patroni/dcs/etcd3.py

@@ -198,12 +198,6 @@ def build_range_request(key: str, range_end: Union[bytes, str, None] = None) ->
     return fields
 
 
-class ReAuthenticateMode(IntEnum):
-    NOT_REQUIRED = 0
-    REQUIRED = 1
-    WITHOUT_WATCHER_RESTART = 2
-
-
 def _handle_auth_errors(func: Callable[..., Any]) -> Any:
     def wrapper(self: 'Etcd3Client', *args: Any, **kwargs: Any) -> Any:
         return self.handle_auth_errors(func, *args, **kwargs)
@@ -215,7 +209,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
     ERROR_CLS = Etcd3Error
 
     def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
-        self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
+        self._reauthenticate = False
         self._token = None
         self._cluster_version: Tuple[int, ...] = tuple()
         super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl)
@@ -294,7 +288,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
         fields['retry'] = retry
         return self.api_execute(self.version_prefix + method, self._MPOST, fields)
 
-    def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool:
+    def authenticate(self, *, retry: Optional[Retry] = None) -> bool:
         if self._use_proxies and not self._cluster_version:
             kwargs = self._prepare_common_parameters(1)
             self._ensure_version_prefix(self._base_uri, **kwargs)
@@ -316,20 +310,18 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
 
     def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any,
                            retry: Optional[Retry] = None, **kwargs: Any) -> Any:
+        reauthenticated = False
         exc = None
         while True:
-            if self._reauthenticate_reason:
+            if self._reauthenticate:
                 if self.username and self.password:
-                    self.authenticate(
-                        restart_watcher=self._reauthenticate_reason != ReAuthenticateMode.WITHOUT_WATCHER_RESTART,
-                        retry=retry)
-                    self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
-                    if retry:
-                        retry.ensure_deadline(0)
+                    self.authenticate(retry=retry)
+                    self._reauthenticate = False
                 else:
                     msg = 'Username or password not set, authentication is not possible'
                     logger.fatal(msg)
                     raise exc or Etcd3Exception(msg)
+                reauthenticated = True
 
             try:
                 return func(self, *args, retry=retry, **kwargs)
@@ -347,11 +339,12 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
             except AuthOldRevision as e:
                 logger.error('Auth token is for old revision of auth store')
                 exc = e
-            self._reauthenticate_reason = ReAuthenticateMode.WITHOUT_WATCHER_RESTART \
-                if isinstance(exc, AuthOldRevision) else ReAuthenticateMode.REQUIRED
-            if not retry:
+            self._reauthenticate = True
+            if retry:
+                logger.error('retry = %s', retry)
+                retry.ensure_deadline(0.5, exc)
+            elif reauthenticated:
                 raise exc
-            retry.ensure_deadline(0.5, exc)
 
     @_handle_auth_errors
     def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True,
@@ -603,12 +596,6 @@ class PatroniEtcd3Client(Etcd3Client):
         super(PatroniEtcd3Client, self).set_base_uri(value)
         self._restart_watcher()
 
-    def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool:
-        ret = super(PatroniEtcd3Client, self).authenticate(restart_watcher=restart_watcher, retry=retry)
-        if ret and restart_watcher:
-            self._restart_watcher()
-        return ret
-
     def _wait_cache(self, timeout: float) -> None:
         stop_time = time.time() + timeout
         while self._kv_cache and not self._kv_cache.is_ready():
@@ -866,14 +853,16 @@ class Etcd3(AbstractEtcd):
         try:
             return _retry(self._client.put, self.leader_path, self._name, self._lease, create_revision='0')
         except LeaseNotFound:
-            logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt')
             self._lease = None
-            retry.ensure_deadline(0)
+            if not retry.ensure_deadline(0):
+                logger.error('Our lease disappeared from Etcd. Deadline exceeded, giving up')
+                return False
+
+            logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt')
 
             _retry(self._do_refresh_lease)
 
             retry.ensure_deadline(1, Etcd3Error('_do_attempt_to_acquire_leader timeout'))
-
             return _retry(self._client.put, self.leader_path, self._name, self._lease, create_revision='0')
 
     @catch_return_false_exception

patroni/dcs/kubernetes.py

@@ -20,6 +20,7 @@ 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, Status, SyncState, TimelineHistory
+from ..collections import EMPTY_DICT
 from ..exceptions import DCSError
 from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \
@@ -470,7 +471,7 @@ class K8sClient(object):
                 if len(args) == 3:  # name, namespace, body
                     body = args[2]
                 elif action == 'create':  # namespace, body
-                    body = args[1]
+                    body = args[1]  # pyright: ignore [reportGeneralTypeIssues]
                 elif action == 'delete':  # name, namespace
                     body = kwargs.pop('body', None)
                 else:
@@ -509,7 +510,7 @@ class KubernetesRetriableException(k8s_client.rest.ApiException):
     @property
     def sleeptime(self) -> Optional[int]:
         try:
-            return int((self.headers or {}).get('retry-after', ''))
+            return int((self.headers or EMPTY_DICT).get('retry-after', ''))
         except Exception:
             return None
 
@@ -654,7 +655,7 @@ class ObjectCache(Thread):
             obj = K8sObject(obj)
             success, old_value = self.set(name, obj)
             if success:
-                new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name))
+                new_value = (obj.metadata.annotations or EMPTY_DICT).get(self._annotations_map.get(name, ''))
         elif ev_type == 'DELETED':
             success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
         else:
@@ -662,7 +663,7 @@ class ObjectCache(Thread):
 
         if success and obj.get('kind') != 'Pod':
             if old_value:
-                old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
+                old_value = (old_value.metadata.annotations or EMPTY_DICT).get(self._annotations_map.get(name, ''))
 
             value_changed = old_value != new_value and \
                 (name != self._dcs.config_path or old_value is not None and new_value is not None)
@@ -844,7 +845,7 @@ class Kubernetes(AbstractDCS):
 
     @staticmethod
     def member(pod: K8sObject) -> Member:
-        annotations = pod.metadata.annotations or {}
+        annotations = pod.metadata.annotations or EMPTY_DICT
         member = Member.from_node(pod.metadata.resource_version, pod.metadata.name, None, annotations.get('status', ''))
         member.data['pod_labels'] = pod.metadata.labels
         return member
@@ -925,7 +926,7 @@ class Kubernetes(AbstractDCS):
         failover = nodes.get(path + self._FAILOVER)
         metadata = failover and failover.metadata
         failover = metadata and Failover.from_node(metadata.resource_version,
-                                                   (metadata.annotations or {}).copy())
+                                                   (metadata.annotations or EMPTY_DICT).copy())
 
         # get synchronization state
         sync = nodes.get(path + self._SYNC)
@@ -1047,8 +1048,9 @@ class Kubernetes(AbstractDCS):
 
     def __target_ref(self, leader_ip: str, latest_subsets: List[K8sObject], pod: K8sObject) -> K8sObject:
         # we want to re-use existing target_ref if possible
+        empty_addresses: List[K8sObject] = []
         for subset in latest_subsets:
-            for address in subset.addresses or []:
+            for address in subset.addresses or empty_addresses:
                 if address.ip == leader_ip and address.target_ref and address.target_ref.name == self._name:
                     return address.target_ref
         return k8s_client.V1ObjectReference(kind='Pod', uid=pod.metadata.uid, namespace=self._namespace,
@@ -1056,7 +1058,8 @@ class Kubernetes(AbstractDCS):
 
     def _map_subsets(self, endpoints: Dict[str, Any], ips: List[str]) -> None:
         leader = self._kinds.get(self.leader_path)
-        latest_subsets = leader and leader.subsets or []
+        empty_addresses: List[K8sObject] = []
+        latest_subsets = leader and leader.subsets or empty_addresses
         if not ips:
             # We want to have subsets empty
             if latest_subsets:
@@ -1212,7 +1215,7 @@ class Kubernetes(AbstractDCS):
         if not retry.ensure_deadline(0.5):
             return False
 
-        kind_annotations = kind and kind.metadata.annotations or {}
+        kind_annotations = kind and kind.metadata.annotations or EMPTY_DICT
         kind_resource_version = kind and kind.metadata.resource_version
 
         # There is different leader or resource_version in cache didn't change
@@ -1225,7 +1228,7 @@ class Kubernetes(AbstractDCS):
     def update_leader(self, leader: Leader, last_lsn: Optional[int],
                       slots: Optional[Dict[str, int]] = None, failsafe: Optional[Dict[str, str]] = None) -> bool:
         kind = self._kinds.get(self.leader_path)
-        kind_annotations = kind and kind.metadata.annotations or {}
+        kind_annotations = kind and kind.metadata.annotations or EMPTY_DICT
 
         if kind and kind_annotations.get(self._LEADER) != self._name:
             return False
@@ -1346,7 +1349,7 @@ class Kubernetes(AbstractDCS):
     def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool:
         ret = False
         kind = self._kinds.get(self.leader_path)
-        if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name:
+        if kind and (kind.metadata.annotations or EMPTY_DICT).get(self._LEADER) == self._name:
             annotations: Dict[str, Optional[str]] = {self._LEADER: None}
             if last_lsn:
                 annotations[self._OPTIME] = str(last_lsn)

patroni/global_config.py

@@ -10,6 +10,7 @@ import types
 from copy import deepcopy
 from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING
 
+from .collections import EMPTY_DICT
 from .utils import parse_bool, parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -44,19 +45,20 @@ class GlobalConfig(types.ModuleType):
         """
         return bool(cluster and cluster.config and cluster.config.modify_version)
 
-    def update(self, cluster: Optional['Cluster']) -> None:
+    def update(self, cluster: Optional['Cluster'], default: Optional[Dict[str, Any]] = None) -> None:
         """Update with the new global configuration from the :class:`Cluster` object view.
 
         .. note::
-            Global configuration is updated only when configuration in the *cluster* view is valid.
-
             Update happens in-place and is executed only from the main heartbeat thread.
 
         :param cluster: the currently known cluster state from DCS.
+        :param default: default configuration, which will be used if there is no valid *cluster.config*.
         """
         # Try to protect from the case when DCS was wiped out
         if self._cluster_has_valid_config(cluster):
             self.__config = cluster.config.data  # pyright: ignore [reportOptionalMemberAccess]
+        elif default:
+            self.__config = default
 
     def from_cluster(self, cluster: Optional['Cluster']) -> 'GlobalConfig':
         """Return :class:`GlobalConfig` instance from the provided :class:`Cluster` object view.
@@ -213,7 +215,7 @@ class GlobalConfig(types.ModuleType):
     @property
     def use_slots(self) -> bool:
         """``True`` if cluster is configured to use replication slots."""
-        return bool(parse_bool((self.get('postgresql') or {}).get('use_slots', True)))
+        return bool(parse_bool((self.get('postgresql') or EMPTY_DICT).get('use_slots', True)))
 
     @property
     def permanent_slots(self) -> Dict[str, Any]:
@@ -221,7 +223,7 @@ class GlobalConfig(types.ModuleType):
         return deepcopy(self.get('permanent_replication_slots')
                         or self.get('permanent_slots')
                         or self.get('slots')
-                        or {})
+                        or EMPTY_DICT.copy())
 
 
 sys.modules[__name__] = GlobalConfig()

patroni/ha.py

@@ -185,6 +185,9 @@ class Ha(object):
         # used only in backoff after failing a pre_promote script
         self._released_leader_key_timestamp = 0
 
+        # Initialize global config
+        global_config.update(None, self.patroni.config.dynamic_configuration)
+
     def primary_stop_timeout(self) -> Union[int, None]:
         """:returns: "primary_stop_timeout" from the global configuration or `None` when not in synchronous mode."""
         ret = global_config.primary_stop_timeout
@@ -607,9 +610,12 @@ class Ha(object):
 
         :returns: the node which we should be replicating from.
         """
+        # nostream is set, the node must not use WAL streaming
+        if self.patroni.nostream:
+            return None
         # 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() \
+        elif self.is_standby_cluster() \
                 and (cluster.leader and cluster.leader.name and cluster.leader.name == self.state_handler.name
                      or cluster.is_unlocked() and not self.is_paused()):
             node_to_follow = self.get_remote_member()

patroni/log.py

@@ -413,7 +413,8 @@ class PatroniLogger(Thread):
                 if not isinstance(handler, RotatingFileHandler):
                     handler = RotatingFileHandler(os.path.join(config['dir'], __name__))
 
-                handler.maxBytes = int(config.get('file_size', 25000000))  # pyright: ignore [reportGeneralTypeIssues]
+                max_file_size = int(config.get('file_size', 25000000))
+                handler.maxBytes = max_file_size  # pyright: ignore [reportAttributeAccessIssue]
                 handler.backupCount = int(config.get('file_num', 4))
             # we can't use `if not isinstance(handler, logging.StreamHandler)` below,
             # because RotatingFileHandler is a child of StreamHandler!!!

patroni/postgresql/__init__.py

@@ -26,7 +26,7 @@ from .slots import SlotsHandler
 from .sync import SyncHandler
 from .. import global_config, psycopg
 from ..async_executor import CriticalTask
-from ..collections import CaseInsensitiveSet, CaseInsensitiveDict
+from ..collections import CaseInsensitiveSet, CaseInsensitiveDict, EMPTY_DICT
 from ..dcs import Cluster, Leader, Member, SLOT_ADVANCE_AVAILABLE_VERSION
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
@@ -272,7 +272,7 @@ class Postgresql(object):
 
         :returns: path to Postgres binary named *cmd*.
         """
-        return os.path.join(self._bin_dir, (self.config.get('bin_name', {}) or {}).get(cmd, cmd))
+        return os.path.join(self._bin_dir, (self.config.get('bin_name', {}) or EMPTY_DICT).get(cmd, cmd))
 
     def pg_ctl(self, cmd: str, *args: str, **kwargs: Any) -> bool:
         """Builds and executes pg_ctl command
@@ -414,7 +414,7 @@ class Postgresql(object):
         return data_directory_is_empty(self._data_dir)
 
     def replica_method_options(self, method: str) -> Dict[str, Any]:
-        return deepcopy(self.config.get(method, {}) or {})
+        return deepcopy(self.config.get(method, {}) or EMPTY_DICT.copy())
 
     def replica_method_can_work_without_replication_connection(self, method: str) -> bool:
         return method != 'basebackup' and bool(self.replica_method_options(method).get('no_master')

patroni/postgresql/available_parameters/__init__.py

@@ -0,0 +1,59 @@
+import logging
+import sys
+
+from typing import Iterator
+
+logger = logging.getLogger(__name__)
+
+if sys.version_info < (3, 9):  # pragma: no cover
+    from pathlib import Path
+
+    PathLikeObj = Path
+    conf_dir = Path(__file__).parent
+else:
+    from importlib.resources import files
+
+    if sys.version_info < (3, 11):  # pragma: no cover
+        from importlib.abc import Traversable
+    else:  # pragma: no cover
+        from importlib.resources.abc import Traversable
+
+    PathLikeObj = Traversable
+    conf_dir = files(__name__)
+
+
+def get_validator_files() -> Iterator[PathLikeObj]:
+    """Recursively find YAML files from the current package directory.
+
+    :returns: an iterator of :class:`PathLikeObj` objects representing validator files.
+    """
+    return _traversable_walk(conf_dir.iterdir())
+
+
+def _traversable_walk(tvbs: Iterator[PathLikeObj]) -> Iterator[PathLikeObj]:
+    """Recursively walk through Path/Traversable objects, yielding all YAML files in deterministic order.
+
+    :param tvbs: An iterator over :class:`PathLikeObj` objects, where each object is a file or directory
+                 that potentially contains YAML files.
+
+    :yields: :class:`PathLikeObj` objects representing YAML files found during the traversal.
+    """
+    for tvb in _filter_and_sort_files(tvbs):
+        if tvb.is_file():
+            yield tvb
+        elif tvb.is_dir():
+            yield from _traversable_walk(tvb.iterdir())
+
+
+def _filter_and_sort_files(files: Iterator[PathLikeObj]) -> Iterator[PathLikeObj]:
+    """Sort files by name, and filter out non-YAML files and Python files.
+
+    :param files: A list of files and/or directories to be filtered and sorted.
+
+    :yields: filtered and sorted objects.
+    """
+    for file in sorted(files, key=lambda x: x.name):
+        if file.name.lower().endswith((".yml", ".yaml")) or file.is_dir():
+            yield file
+        elif not file.name.lower().endswith((".py", ".pyc")):
+            logger.info("Ignored a non-YAML file found under `%s` directory: `%s`.", __name__.split('.')[-1], file)

patroni/postgresql/bootstrap.py

@@ -7,6 +7,7 @@ import time
 from typing import Any, Callable, Dict, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from ..async_executor import CriticalTask
+from ..collections import EMPTY_DICT
 from ..dcs import Leader, Member, RemoteMember
 from ..psycopg import quote_ident, quote_literal
 from ..utils import deep_compare, unquote
@@ -146,7 +147,7 @@ class Bootstrap(object):
 
         # make sure there is no trigger file or postgres will be automatically promoted
         trigger_file = self._postgresql.config.triggerfile_good_name
-        trigger_file = (self._postgresql.config.get('recovery_conf') or {}).get(trigger_file) or 'promote'
+        trigger_file = (self._postgresql.config.get('recovery_conf') or EMPTY_DICT).get(trigger_file) or 'promote'
         trigger_file = os.path.abspath(os.path.join(self._postgresql.data_dir, trigger_file))
         if os.path.exists(trigger_file):
             os.unlink(trigger_file)
@@ -441,7 +442,7 @@ END;$$""".format(f, quote_ident(rewind['username'], postgresql.connection()))
                 if config.get('users'):
                     logger.warning('User creation via "bootstrap.users" will be removed in v4.0.0')
 
-                for name, value in (config.get('users') or {}).items():
+                for name, value in (config.get('users') or EMPTY_DICT).items():
                     if all(name != a.get('username') for a in (superuser, replication, rewind)):
                         self.create_or_update_role(name, value.get('password'), value.get('options', []))
 

patroni/postgresql/cancellable.py

@@ -100,7 +100,8 @@ class CancellableSubprocess(CancellableExecutor):
 
             if started and self._process is not None:
                 if isinstance(communicate, dict):
-                    communicate['stdout'], communicate['stderr'] = self._process.communicate(input_data)
+                    communicate['stdout'], communicate['stderr'] = \
+                        self._process.communicate(input_data)  # pyright: ignore [reportGeneralTypeIssues]
                 return self._process.wait()
         finally:
             with self._lock:

patroni/postgresql/config.py

@@ -13,7 +13,7 @@ from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Un
 
 from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value
 from .. import global_config
-from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
+from ..collections import CaseInsensitiveDict, CaseInsensitiveSet, EMPTY_DICT
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException, PostgresConnectionException
 from ..file_perm import pg_perm
@@ -619,7 +619,8 @@ class ConfigHandler(object):
             fd.write_param(name, value)
 
     def build_recovery_params(self, member: Union[Leader, Member, None]) -> CaseInsensitiveDict:
-        recovery_params = CaseInsensitiveDict({p: v for p, v in (self.get('recovery_conf') or {}).items()
+        default: Dict[str, Any] = {}
+        recovery_params = CaseInsensitiveDict({p: v for p, v in (self.get('recovery_conf') or default).items()
                                                if not p.lower().startswith('recovery_target')
                                                and p.lower() not in ('primary_conninfo', 'primary_slot_name')})
         recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'})
@@ -639,8 +640,7 @@ class ConfigHandler(object):
                 # We are a standby leader and are using a replication slot. Make sure we connect to
                 # the leader of the main cluster (in case more than one host is specified in the
                 # connstr) by adding 'target_session_attrs=read-write' to primary_conninfo.
-                if is_remote_member and 'target_sesions_attrs' not in primary_conninfo and\
-                        self._postgresql.major_version >= 100000:
+                if is_remote_member and ',' in primary_conninfo['host'] and self._postgresql.major_version >= 100000:
                     primary_conninfo['target_session_attrs'] = 'read-write'
             recovery_params['primary_conninfo'] = primary_conninfo
 
@@ -846,7 +846,7 @@ class ConfigHandler(object):
             required['restart' if mtype else 'reload'] += 1
 
         wanted_recovery_params = self.build_recovery_params(member)
-        for param, value in (self._current_recovery_params or {}).items():
+        for param, value in (self._current_recovery_params or EMPTY_DICT).items():
             # Skip certain parameters defined in the included postgres config files
             # if we know that they are not specified in the patroni configuration.
             if len(value) > 2 and value[2] not in (self._postgresql_conf, self._auto_conf) and \
@@ -1325,4 +1325,4 @@ class ConfigHandler(object):
         return self._config.get(key, default)
 
     def restore_command(self) -> Optional[str]:
-        return (self.get('recovery_conf') or {}).get('restore_command')
+        return (self.get('recovery_conf') or EMPTY_DICT).get('restore_command')

patroni/postgresql/mpp/__init__.py

@@ -299,6 +299,8 @@ def iter_mpp_classes(
 
     :yields: tuples, each containing the module ``name`` and the imported MPP class object.
     """
+    if TYPE_CHECKING:  # pragma: no cover
+        assert isinstance(__package__, str)
     yield from iter_classes(__package__, AbstractMPP, config)
 
 

patroni/postgresql/rewind.py

@@ -13,6 +13,7 @@ from . import Postgresql
 from .connection import get_connection_cursor
 from .misc import format_lsn, fsync_dir, parse_history, parse_lsn
 from ..async_executor import CriticalTask
+from ..collections import EMPTY_DICT
 from ..dcs import Leader, RemoteMember
 
 logger = logging.getLogger(__name__)
@@ -209,9 +210,10 @@ class Rewind(object):
         ret = member.conn_kwargs(auth)
         if not ret.get('dbname'):
             ret['dbname'] = self._postgresql.database
-        # Add target_session_attrs in case more than one hostname is specified
-        # (libpq client-side failover) making sure we hit the primary
-        if 'target_session_attrs' not in ret and self._postgresql.major_version >= 100000:
+        # Add target_session_attrs to make sure we hit the primary.
+        # It is not strictly necessary for starting from PostgreSQL v14, which made it possible
+        # to rewind from standby, but doing it from the real primary is always safer.
+        if self._postgresql.major_version >= 100000:
             ret['target_session_attrs'] = 'read-write'
         return ret
 
@@ -417,7 +419,7 @@ class Rewind(object):
         dsn = self._postgresql.config.format_dsn(r, True)
         logger.info('running pg_rewind from %s', dsn)
 
-        restore_command = (self._postgresql.config.get('recovery_conf') or {}).get('restore_command') \
+        restore_command = (self._postgresql.config.get('recovery_conf') or EMPTY_DICT).get('restore_command') \
             if self._postgresql.major_version < 120000 else self._postgresql.get_guc_value('restore_command')
 
         # Until v15 pg_rewind expected postgresql.conf to be inside $PGDATA, which is not the case on e.g. Debian

patroni/postgresql/validator.py

@@ -1,11 +1,11 @@
 import abc
 from copy import deepcopy
 import logging
-import os
 import yaml
 
 from typing import Any, Dict, Iterator, List, MutableMapping, Optional, Tuple, Type, Union
 
+from .available_parameters import get_validator_files, PathLikeObj
 from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..exceptions import PatroniException
 from ..utils import parse_bool, parse_int, parse_real
@@ -258,10 +258,10 @@ class InvalidGucValidatorsFile(PatroniException):
     """Raised when reading or parsing of a YAML file faces an issue."""
 
 
-def _read_postgres_gucs_validators_file(file: str) -> Dict[str, Any]:
+def _read_postgres_gucs_validators_file(file: PathLikeObj) -> Dict[str, Any]:
     """Read an YAML file and return the corresponding Python object.
 
-    :param file: path to the file to be read. It is expected to be encoded with ``UTF-8``, and to be a YAML document.
+    :param file: path-like object to read from. It is expected to be encoded with ``UTF-8``, and to be a YAML document.
 
     :returns: the YAML content parsed into a Python object. If any issue is faced while reading/parsing the file, then
         return ``None``.
@@ -270,7 +270,7 @@ def _read_postgres_gucs_validators_file(file: str) -> Dict[str, Any]:
         :class:`InvalidGucValidatorsFile`: if faces an issue while reading or parsing *file*.
     """
     try:
-        with open(file, encoding='UTF-8') as stream:
+        with file.open(encoding='UTF-8') as stream:
             return yaml.safe_load(stream)
     except Exception as exc:
         raise InvalidGucValidatorsFile(
@@ -385,21 +385,7 @@ def _load_postgres_gucs_validators() -> None:
                     version_till: null
 
     """
-    conf_dir = os.path.join(
-        os.path.dirname(os.path.abspath(__file__)),
-        'available_parameters',
-    )
-    yaml_files: List[str] = []
-
-    for root, _, files in os.walk(conf_dir):
-        for file in sorted(files):
-            full_path = os.path.join(root, file)
-            if file.lower().endswith(('.yml', '.yaml')):
-                yaml_files.append(full_path)
-            else:
-                logger.info('Ignored a non-YAML file found under `available_parameters` directory: `%s`.', full_path)
-
-    for file in yaml_files:
+    for file in get_validator_files():
         try:
             config: Dict[str, Any] = _read_postgres_gucs_validators_file(file)
         except InvalidGucValidatorsFile as exc:

patroni/psycopg.py

@@ -42,7 +42,8 @@ try:
             value.prepare(conn)
         return value.getquoted().decode('utf-8')
 except ImportError:
-    from psycopg import connect as __connect, sql, Error, DatabaseError, OperationalError, ProgrammingError
+    from psycopg import connect as __connect  # pyright: ignore [reportUnknownVariableType]
+    from psycopg import sql, Error, DatabaseError, OperationalError, ProgrammingError
 
     def _connect(dsn: Optional[str] = None, **kwargs: Any) -> 'Connection[Any]':
         """Call :func:`psycopg.connect` with *dsn* and ``**kwargs``.
@@ -56,7 +57,7 @@ except ImportError:
 
         :returns: a connection to the database.
         """
-        ret = __connect(dsn or "", **kwargs)
+        ret: 'Connection[Any]' = __connect(dsn or "", **kwargs)
         setattr(ret, 'server_version', ret.pgconn.server_version)  # compatibility with psycopg2
         return ret
 

patroni/scripts/barman/__init__.py

@@ -0,0 +1 @@
+"""Create :mod:`patroni.scripts.barman`."""

patroni/scripts/barman/cli.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python
+
+"""Perform operations on Barman through ``pg-backup-api``.
+
+The actual operations are implemented by separate modules. This module only
+builds the CLI that makes an interface with the actual commands.
+
+.. note::
+    See :class:ExitCode` for possible exit codes of this main script.
+"""
+
+from argparse import ArgumentParser
+from enum import IntEnum
+import logging
+import sys
+
+from .config_switch import run_barman_config_switch
+from .recover import run_barman_recover
+from .utils import ApiNotOk, PgBackupApi, set_up_logging
+
+
+class ExitCode(IntEnum):
+    """Possible exit codes of this script.
+
+    :cvar NO_COMMAND: if no sub-command of ``patroni_barman`` application has
+        been selected by the user.
+    :cvar API_NOT_OK: ``pg-backup-api`` status is not ``OK``.
+    """
+
+    NO_COMMAND = -1
+    API_NOT_OK = -2
+
+
+def main() -> None:
+    """Entry point of ``patroni_barman`` application.
+
+    Implements the parser for the application and for its sub-commands.
+
+    The script exit code may be one of:
+
+    * :attr:`ExitCode.NO_COMMAND`: if no sub-command was specified in the
+        ``patroni_barman`` call;
+    * :attr:`ExitCode.API_NOT_OK`: if ``pg-backup-api`` is not correctly up and
+        running;
+    * Value returned by :func:`~patroni.scripts.barman.config_switch.run_barman_config_switch`,
+        if running ``patroni_barman config-switch``;
+    * Value returned by :func:`~patroni.scripts.barman.recover.run_barman_recover`,
+        if running ``patroni_barman recover``.
+
+    The called sub-command is expected to exit execution once finished using
+    its own set of exit codes.
+    """
+    parser = ArgumentParser(
+        description=(
+            "Wrapper application for pg-backup-api. Communicate with the API "
+            "running at the given URL to perform remote Barman operations."
+        ),
+    )
+    parser.add_argument(
+        "--api-url",
+        type=str,
+        required=True,
+        help="URL to reach the pg-backup-api, e.g. 'http://localhost:7480'",
+        dest="api_url",
+    )
+    parser.add_argument(
+        "--cert-file",
+        type=str,
+        required=False,
+        help="Certificate to authenticate against the API, if required.",
+        dest="cert_file",
+    )
+    parser.add_argument(
+        "--key-file",
+        type=str,
+        required=False,
+        help="Certificate key to authenticate against the API, if required.",
+        dest="key_file",
+    )
+    parser.add_argument(
+        "--retry-wait",
+        type=int,
+        required=False,
+        default=2,
+        help="How long in seconds to wait before retrying a failed "
+             "pg-backup-api request (default: '%(default)s')",
+        dest="retry_wait",
+    )
+    parser.add_argument(
+        "--max-retries",
+        type=int,
+        required=False,
+        default=5,
+        help="Maximum number of retries when receiving malformed responses "
+             "from the pg-backup-api (default: '%(default)s')",
+        dest="max_retries",
+    )
+    parser.add_argument(
+        "--log-file",
+        type=str,
+        required=False,
+        help="File where to log messages produced by this application, if any.",
+        dest="log_file",
+    )
+
+    subparsers = parser.add_subparsers(title="Sub-commands")
+
+    recover_parser = subparsers.add_parser(
+        "recover",
+        help="Remote 'barman recover'",
+        description="Restore a Barman backup of a given Barman server"
+    )
+    recover_parser.add_argument(
+        "--barman-server",
+        type=str,
+        required=True,
+        help="Name of the Barman server from which to restore the backup.",
+        dest="barman_server",
+    )
+    recover_parser.add_argument(
+        "--backup-id",
+        type=str,
+        required=False,
+        default="latest",
+        help="ID of the Barman backup to be restored. You can use any value "
+             "supported by 'barman recover' command "
+             "(default: '%(default)s')",
+        dest="backup_id",
+    )
+    recover_parser.add_argument(
+        "--ssh-command",
+        type=str,
+        required=True,
+        help="Value to be passed as '--remote-ssh-command' to 'barman recover'.",
+        dest="ssh_command",
+    )
+    recover_parser.add_argument(
+        "--data-directory",
+        "--datadir",
+        type=str,
+        required=True,
+        help="Destination path where to restore the barman backup in the "
+             "local host.",
+        dest="data_directory",
+    )
+    recover_parser.add_argument(
+        "--loop-wait",
+        type=int,
+        required=False,
+        default=10,
+        help="How long to wait before checking again the status of the "
+             "recovery process, in seconds. Use higher values if your "
+             "recovery is expected to take long (default: '%(default)s')",
+        dest="loop_wait",
+    )
+    recover_parser.set_defaults(func=run_barman_recover)
+
+    config_switch_parser = subparsers.add_parser(
+        "config-switch",
+        help="Remote 'barman config-switch'",
+        description="Switch the configuration of a given Barman server. "
+                    "Intended to be used as a 'on_role_change' callback."
+    )
+    config_switch_parser.add_argument(
+        "action",
+        type=str,
+        choices=["on_role_change"],
+        help="Name of the callback (automatically filled by Patroni)",
+    )
+    config_switch_parser.add_argument(
+        "role",
+        type=str,
+        choices=["master", "primary", "promoted", "standby_leader", "replica",
+                 "demoted"],
+        help="Name of the new role of this node (automatically filled by "
+             "Patroni)",
+    )
+    config_switch_parser.add_argument(
+        "cluster",
+        type=str,
+        help="Name of the Patroni cluster involved in the callback "
+             "(automatically filled by Patroni)",
+    )
+    config_switch_parser.add_argument(
+        "--barman-server",
+        type=str,
+        required=True,
+        help="Name of the Barman server which config is to be switched.",
+        dest="barman_server",
+    )
+    group = config_switch_parser.add_mutually_exclusive_group(required=True)
+    group.add_argument(
+        "--barman-model",
+        type=str,
+        help="Name of the Barman config model to be applied to the server.",
+        dest="barman_model",
+    )
+    group.add_argument(
+        "--reset",
+        action="store_true",
+        help="Unapply the currently active model for the server, if any.",
+        dest="reset",
+    )
+    config_switch_parser.add_argument(
+        "--switch-when",
+        type=str,
+        required=True,
+        default="promoted",
+        choices=["promoted", "demoted", "always"],
+        help="Controls under which circumstances the 'on_role_change' callback "
+             "should actually switch config in Barman. 'promoted' means the "
+             "'role' is either 'master', 'primary' or 'promoted'. 'demoted' "
+             "means the 'role' is either 'replica' or 'demoted' "
+             "(default: '%(default)s')",
+        dest="switch_when",
+    )
+    config_switch_parser.set_defaults(func=run_barman_config_switch)
+
+    args, _ = parser.parse_known_args()
+
+    set_up_logging(args.log_file)
+
+    if not hasattr(args, "func"):
+        parser.print_help()
+        sys.exit(ExitCode.NO_COMMAND)
+
+    api = None
+
+    try:
+        api = PgBackupApi(args.api_url, args.cert_file, args.key_file,
+                          args.retry_wait, args.max_retries)
+    except ApiNotOk as exc:
+        logging.error("pg-backup-api is not working: %r", exc)
+        sys.exit(ExitCode.API_NOT_OK)
+
+    sys.exit(args.func(api, args))
+
+
+if __name__ == "__main__":
+    main()

patroni/scripts/barman/config_switch.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+
+"""Implements ``patroni_barman config-switch`` sub-command.
+
+Apply a Barman configuration model through ``pg-backup-api``.
+
+This sub-command is specially useful as a ``on_role_change`` callback to change
+Barman configuration in response to failovers and switchovers. Check the output
+of ``--help`` to understand the parameters supported by the sub-command.
+
+It requires that you have previously configured a Barman server and Barman
+config models, and that you have ``pg-backup-api`` configured and running in
+the same host as Barman.
+
+Refer to :class:`ExitCode` for possible exit codes of this sub-command.
+"""
+from argparse import Namespace
+from enum import IntEnum
+import logging
+import time
+from typing import Optional, TYPE_CHECKING
+
+from .utils import OperationStatus, RetriesExceeded
+
+
+if TYPE_CHECKING:  # pragma: no cover
+    from .utils import PgBackupApi
+
+
+class ExitCode(IntEnum):
+    """Possible exit codes of this script.
+
+    :cvar CONFIG_SWITCH_DONE: config switch was successfully performed.
+    :cvar CONFIG_SWITCH_SKIPPED: if the execution was skipped because of not
+        matching user expectations.
+    :cvar CONFIG_SWITCH_FAILED: config switch faced an issue.
+    :cvar HTTP_ERROR: an error has occurred while communicating with
+        ``pg-backup-api``
+    :cvar INVALID_ARGS: an invalid set of arguments has been given to the
+        operation.
+    """
+
+    CONFIG_SWITCH_DONE = 0
+    CONFIG_SWITCH_SKIPPED = 1
+    CONFIG_SWITCH_FAILED = 2
+    HTTP_ERROR = 3
+    INVALID_ARGS = 4
+
+
+def _should_skip_switch(args: Namespace) -> bool:
+    """Check if we should skip the config switch operation.
+
+    :param args: arguments received from the command-line of
+        ``patroni_barman config-switch`` command.
+
+    :returns: if the operation should be skipped.
+    """
+    if args.switch_when == "promoted":
+        return args.role not in {"master", "primary", "promoted"}
+    if args.switch_when == "demoted":
+        return args.role not in {"replica", "demoted"}
+    return False
+
+
+def _switch_config(api: "PgBackupApi", barman_server: str,
+                   barman_model: Optional[str], reset: Optional[bool]) -> int:
+    """Switch configuration of Barman server through ``pg-backup-api``.
+
+    .. note::
+        If requests to ``pg-backup-api`` fail recurrently or we face HTTP
+        errors, then exit with :attr:`ExitCode.HTTP_ERROR`.
+
+    :param api: a :class:`PgBackupApi` instance to handle communication with
+        the API.
+    :param barman_server: name of the Barman server which config is to be
+        switched.
+    :param barman_model: name of the Barman model to be applied to the server,
+        if any.
+    :param reset: ``True`` if you would like to unapply the currently active
+        model for the server, if any.
+
+    :returns: the return code to be used when exiting the ``patroni_barman``
+        application. Refer to :class:`ExitCode`.
+    """
+    operation_id = None
+
+    try:
+        operation_id = api.create_config_switch_operation(
+            barman_server,
+            barman_model,
+            reset,
+        )
+    except RetriesExceeded as exc:
+        logging.error("An issue was faced while trying to create a config "
+                      "switch operation: %r", exc)
+        return ExitCode.HTTP_ERROR
+
+    logging.info("Created the config switch operation with ID %s",
+                 operation_id)
+
+    status = None
+
+    while True:
+        try:
+            status = api.get_operation_status(barman_server, operation_id)
+        except RetriesExceeded:
+            logging.error("Maximum number of retries exceeded, exiting.")
+            return ExitCode.HTTP_ERROR
+
+        if status != OperationStatus.IN_PROGRESS:
+            break
+
+        logging.info("Config switch operation %s is still in progress",
+                     operation_id)
+        time.sleep(5)
+
+    if status == OperationStatus.DONE:
+        logging.info("Config switch operation finished successfully.")
+        return ExitCode.CONFIG_SWITCH_DONE
+    else:
+        logging.error("Config switch operation failed.")
+        return ExitCode.CONFIG_SWITCH_FAILED
+
+
+def run_barman_config_switch(api: "PgBackupApi", args: Namespace) -> int:
+    """Run a remote ``barman config-switch`` through the ``pg-backup-api``.
+
+    :param api: a :class:`PgBackupApi` instance to handle communication with
+        the API.
+    :param args: arguments received from the command-line of
+        ``patroni_barman config-switch`` command.
+
+    :returns: the return code to be used when exiting the ``patroni_barman``
+        application. Refer to :class:`ExitCode`.
+    """
+    if _should_skip_switch(args):
+        logging.info("Config switch operation was skipped (role=%s, "
+                     "switch_when=%s).", args.role, args.switch_when)
+        return ExitCode.CONFIG_SWITCH_SKIPPED
+
+    if not bool(args.barman_model) ^ bool(args.reset):
+        logging.error("One, and only one among 'barman_model' ('%s') and "
+                      "'reset' ('%s') should be given", args.barman_model, args.reset)
+        return ExitCode.INVALID_ARGS
+
+    return _switch_config(api, args.barman_server, args.barman_model, args.reset)

patroni/scripts/barman/recover.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python
+
+"""Implements ``patroni_barman recover`` sub-command.
+
+Restore a Barman backup to the local node through ``pg-backup-api``.
+
+This sub-command can be used both as a custom bootstrap method, and as a custom
+create replica method. Check the output of ``--help`` to understand the
+parameters supported by the sub-command. ``--datadir`` is a special parameter
+and it is automatically filled by Patroni in both cases.
+
+It requires that you have previously configured a Barman server, and that you
+have ``pg-backup-api`` configured and running in the same host as Barman.
+
+Refer to :class:`ExitCode` for possible exit codes of this sub-command.
+"""
+from argparse import Namespace
+from enum import IntEnum
+import logging
+import time
+from typing import TYPE_CHECKING
+
+from .utils import OperationStatus, RetriesExceeded
+
+
+if TYPE_CHECKING:  # pragma: no cover
+    from .utils import PgBackupApi
+
+
+class ExitCode(IntEnum):
+    """Possible exit codes of this script.
+
+    :cvar RECOVERY_DONE: backup was successfully restored.
+    :cvar RECOVERY_FAILED: recovery of the backup faced an issue.
+    :cvar HTTP_ERROR: an error has occurred while communicating with
+        ``pg-backup-api``
+    """
+
+    RECOVERY_DONE = 0
+    RECOVERY_FAILED = 1
+    HTTP_ERROR = 2
+
+
+def _restore_backup(api: "PgBackupApi", barman_server: str, backup_id: str,
+                    ssh_command: str, data_directory: str,
+                    loop_wait: int) -> int:
+    """Restore the configured Barman backup through ``pg-backup-api``.
+
+    .. note::
+        If requests to ``pg-backup-api`` fail recurrently or we face HTTP
+        errors, then exit with :attr:`ExitCode.HTTP_ERROR`.
+
+    :param api: a :class:`PgBackupApi` instance to handle communication with
+        the API.
+    :param barman_server: name of the Barman server which backup is to be
+        restored.
+    :param backup_id: ID of the backup from the Barman server.
+    :param ssh_command: SSH command to connect from the Barman host to the
+        target host.
+    :param data_directory: path to the Postgres data directory where to restore
+        the backup in.
+    :param loop_wait: how long in seconds to wait before checking again the
+        status of the recovery process. Higher values are useful for backups
+        that are expected to take longer to restore.
+
+    :returns: the return code to be used when exiting the ``patroni_barman``
+        application. Refer to :class:`ExitCode`.
+    """
+    operation_id = None
+
+    try:
+        operation_id = api.create_recovery_operation(
+            barman_server,
+            backup_id,
+            ssh_command,
+            data_directory,
+        )
+    except RetriesExceeded as exc:
+        logging.error("An issue was faced while trying to create a recovery "
+                      "operation: %r", exc)
+        return ExitCode.HTTP_ERROR
+
+    logging.info("Created the recovery operation with ID %s", operation_id)
+
+    status = None
+
+    while True:
+        try:
+            status = api.get_operation_status(barman_server, operation_id)
+        except RetriesExceeded:
+            logging.error("Maximum number of retries exceeded, exiting.")
+            return ExitCode.HTTP_ERROR
+
+        if status != OperationStatus.IN_PROGRESS:
+            break
+
+        logging.info("Recovery operation %s is still in progress",
+                     operation_id)
+        time.sleep(loop_wait)
+
+    if status == OperationStatus.DONE:
+        logging.info("Recovery operation finished successfully.")
+        return ExitCode.RECOVERY_DONE
+    else:
+        logging.error("Recovery operation failed.")
+        return ExitCode.RECOVERY_FAILED
+
+
+def run_barman_recover(api: "PgBackupApi", args: Namespace) -> int:
+    """Run a remote ``barman recover`` through the ``pg-backup-api``.
+
+    :param api: a :class:`PgBackupApi` instance to handle communication with
+        the API.
+    :param args: arguments received from the command-line of
+        ``patroni_barman recover`` command.
+
+    :returns: the return code to be used when exiting the ``patroni_barman``
+        application. Refer to :class:`ExitCode`.
+    """
+    return _restore_backup(api, args.barman_server, args.backup_id,
+                           args.ssh_command, args.data_directory,
+                           args.loop_wait)

patroni/scripts/barman/utils.py

@@ -0,0 +1,308 @@
+#!/usr/bin/env python
+
+"""Utilitary stuff to be used by Barman related scripts."""
+
+from enum import IntEnum
+import json
+import logging
+from typing import Any, Callable, Dict, Optional, Tuple, Type, Union
+import time
+from urllib.parse import urljoin
+
+from urllib3 import PoolManager
+from urllib3.exceptions import MaxRetryError
+from urllib3.response import HTTPResponse
+
+
+class RetriesExceeded(Exception):
+    """Maximum number of retries exceeded."""
+
+
+def retry(exceptions: Union[Type[Exception], Tuple[Type[Exception], ...]]) \
+        -> Any:
+    """Retry an operation n times if expected *exceptions* are faced.
+
+    .. note::
+        Should be used as a decorator of a class' method as it expects the
+        first argument to be a class instance.
+
+        The class which method is going to be decorated should contain a couple
+        attributes:
+
+        * ``max_retries``: maximum retry attempts before failing;
+        * ``retry_wait``: how long in seconds to wait before retrying.
+
+    :param exceptions: exceptions that could trigger a retry attempt.
+
+    :raises:
+        :exc:`RetriesExceeded`: if the maximum number of attempts has been
+            exhausted.
+    """
+    def decorator(func: Callable[..., Any]) -> Any:
+        def inner_func(instance: object, *args: Any, **kwargs: Any) -> Any:
+            times: int = getattr(instance, "max_retries")
+            retry_wait: int = getattr(instance, "retry_wait")
+            method_name = f"{instance.__class__.__name__}.{func.__name__}"
+
+            attempt = 1
+
+            while attempt <= times:
+                try:
+                    return func(instance, *args, **kwargs)
+                except exceptions as exc:
+                    logging.warning("Attempt %d of %d on method %s failed "
+                                    "with %r.",
+                                    attempt, times, method_name, exc)
+                    attempt += 1
+
+                time.sleep(retry_wait)
+
+            raise RetriesExceeded("Maximum number of retries exceeded for "
+                                  f"method {method_name}.")
+        return inner_func
+    return decorator
+
+
+def set_up_logging(log_file: Optional[str] = None) -> None:
+    """Set up logging to file, if *log_file* is given, otherwise to console.
+
+    :param log_file: file where to log messages, if any.
+    """
+    logging.basicConfig(filename=log_file, level=logging.INFO,
+                        format="%(asctime)s %(levelname)s: %(message)s")
+
+
+class OperationStatus(IntEnum):
+    """Possible status of ``pg-backup-api`` operations.
+
+    :cvar IN_PROGRESS: the operation is still ongoing.
+    :cvar FAILED: the operation failed.
+    :cvar DONE: the operation finished successfully.
+    """
+
+    IN_PROGRESS = 0
+    FAILED = 1
+    DONE = 2
+
+
+class ApiNotOk(Exception):
+    """The ``pg-backup-api`` is not currently up and running."""
+
+
+class PgBackupApi:
+    """Facilities for communicating with the ``pg-backup-api``.
+
+    :ivar api_url: base URL to reach the ``pg-backup-api``.
+    :ivar cert_file: certificate to authenticate against the ``pg-backup-api``,
+        if required.
+    :ivar key_file: certificate key to authenticate against the
+        ``pg-backup-api``, if required.
+    :ivar retry_wait: how long in seconds to wait before retrying a failed
+        request to the ``pg-backup-api``.
+    :ivar max_retries: maximum number of retries when ``pg-backup-api`` returns
+        malformed responses.
+    :ivar http: a HTTP pool manager for performing web requests.
+    """
+
+    def __init__(self, api_url: str, cert_file: Optional[str],
+                 key_file: Optional[str], retry_wait: int,
+                 max_retries: int) -> None:
+        """Create a new instance of :class:`BarmanRecover`.
+
+        Make sure the ``pg-backup-api`` is reachable and running fine.
+
+        .. note::
+            When using any method which send requests to the API, be aware that
+            they might raise :exc:`RetriesExceeded` upon HTTP request errors.
+
+            Similarly, when instantiating this class you may face an
+            :exc:`ApiNotOk`, if the API is down or returns a bogus status.
+
+        :param api_url: base URL to reach the ``pg-backup-api``.
+        :param cert_file: certificate to authenticate against the
+            ``pg-backup-api``, if required.
+        :param key_file: certificate key to authenticate against the
+            ``pg-backup-api``, if required.
+        :param retry_wait: how long in seconds to wait before retrying a failed
+            request to the ``pg-backup-api``.
+        :param max_retries: maximum number of retries when ``pg-backup-api``
+            returns malformed responses.
+        """
+        self.api_url = api_url
+        self.cert_file = cert_file
+        self.key_file = key_file
+        self.retry_wait = retry_wait
+        self.max_retries = max_retries
+        self._http = PoolManager(cert_file=cert_file, key_file=key_file)
+        self._ensure_api_ok()
+
+    def _build_full_url(self, url_path: str) -> str:
+        """Build the full URL by concatenating *url_path* with the base URL.
+
+        :param url_path: path to be accessed in the ``pg-backup-api``.
+
+        :returns: the full URL after concatenating.
+        """
+        return urljoin(self.api_url, url_path)
+
+    @staticmethod
+    def _deserialize_response(response: HTTPResponse) -> Any:
+        """Retrieve body from *response* as a deserialized JSON object.
+
+        :param response: response from which JSON body will be deserialized.
+
+        :returns: the deserialized JSON body.
+        """
+        return json.loads(response.data.decode("utf-8"))
+
+    @staticmethod
+    def _serialize_request(body: Any) -> Any:
+        """Serialize a request body.
+
+        :param body: content of the request body to be serialized.
+
+        :returns: the serialized request body.
+        """
+        return json.dumps(body).encode("utf-8")
+
+    def _get_request(self, url_path: str) -> Any:
+        """Perform a ``GET`` request to *url_path*.
+
+        :param url_path: URL to perform the ``GET`` request against.
+
+        :returns: the deserialized response body.
+
+        :raises:
+            :exc:`RetriesExceeded`: raised from the corresponding :mod:`urllib3`
+                exception.
+        """
+        url = self._build_full_url(url_path)
+        response = None
+
+        try:
+            response = self._http.request("GET", url)
+        except MaxRetryError as exc:
+            msg = f"Failed to perform a GET request to {url}"
+            raise RetriesExceeded(msg) from exc
+
+        return self._deserialize_response(response)
+
+    def _post_request(self, url_path: str, body: Any) -> Any:
+        """Perform a ``POST`` request to *url_path* serializing *body* as JSON.
+
+        :param url_path: URL to perform the ``POST`` request against.
+        :param body: the body to be serialized as JSON and sent in the request.
+
+        :returns: the deserialized response body.
+
+        :raises:
+            :exc:`RetriesExceeded`: raised from the corresponding :mod:`urllib3`
+                exception.
+        """
+        body = self._serialize_request(body)
+
+        url = self._build_full_url(url_path)
+        response = None
+
+        try:
+            response = self._http.request("POST",
+                                          url,
+                                          body=body,
+                                          headers={
+                                              "Content-Type": "application/json"
+                                          })
+        except MaxRetryError as exc:
+            msg = f"Failed to perform a POST request to {url} with {body}"
+            raise RetriesExceeded(msg) from exc
+
+        return self._deserialize_response(response)
+
+    def _ensure_api_ok(self) -> None:
+        """Ensure ``pg-backup-api`` is reachable and ``OK``.
+
+        :raises:
+            :exc:`ApiNotOk`: if ``pg-backup-api`` status is not ``OK``.
+        """
+        response = self._get_request("status")
+
+        if response != "OK":
+            msg = (
+                "pg-backup-api is currently not up and running at "
+                f"{self.api_url}: {response}"
+            )
+
+            raise ApiNotOk(msg)
+
+    @retry(KeyError)
+    def get_operation_status(self, barman_server: str,
+                             operation_id: str) -> OperationStatus:
+        """Get status of the operation which ID is *operation_id*.
+
+        :param barman_server: name of the Barman server related with the
+            operation.
+        :param operation_id: ID of the operation to be checked.
+
+        :returns: the status of the operation.
+        """
+        response = self._get_request(
+            f"servers/{barman_server}/operations/{operation_id}",
+        )
+
+        status = response["status"]
+        return OperationStatus[status]
+
+    @retry(KeyError)
+    def create_recovery_operation(self, barman_server: str, backup_id: str,
+                                  ssh_command: str, data_directory: str) -> str:
+        """Create a recovery operation on the ``pg-backup-api``.
+
+        :param barman_server: name of the Barman server which backup is to be
+            restored.
+        :param backup_id: ID of the backup from the Barman server.
+        :param ssh_command: SSH command to connect from the Barman host to the
+            target host.
+        :param data_directory: path to the Postgres data directory where to
+            restore the backup at.
+
+        :returns: the ID of the recovery operation that has been created.
+        """
+        response = self._post_request(
+            f"servers/{barman_server}/operations",
+            {
+                "type": "recovery",
+                "backup_id": backup_id,
+                "remote_ssh_command": ssh_command,
+                "destination_directory": data_directory,
+            },
+        )
+
+        return response["operation_id"]
+
+    @retry(KeyError)
+    def create_config_switch_operation(self, barman_server: str,
+                                       barman_model: Optional[str],
+                                       reset: Optional[bool]) -> str:
+        """Create a config switch operation on the ``pg-backup-api``.
+
+        :param barman_server: name of the Barman server which config is to be
+            switched.
+        :param barman_model: name of the Barman model to be applied to the
+            server, if any.
+        :param reset: ``True`` if you would like to unapply the currently active
+            model for the server, if any.
+
+        :returns: the ID of the config switch operation that has been created.
+        """
+        body: Dict[str, Any] = {"type": "config_switch"}
+
+        if barman_model:
+            body["model_name"] = barman_model
+        elif reset:
+            body["reset"] = reset
+
+        response = self._post_request(
+            f"servers/{barman_server}/operations",
+            body,
+        )
+
+        return response["operation_id"]

patroni/scripts/barman_recover.py

@@ -1,468 +0,0 @@
-#!/usr/bin/env python
-
-"""Restore a Barman backup to the local node through ``pg-backup-api``.
-
-This script can be used both as a custom bootstrap method, and as a custom
-create replica method. Check the output of ``--help`` to understand the
-parameters supported by the script. ``--datadir`` is a special parameter and it
-is automatically filled by Patroni in both cases.
-
-It requires that you have previously configured a Barman server, and that you
-have ``pg-backup-api`` configured and running in the same host as Barman.
-
-Refer to :class:`ExitCode` for possible exit codes of this script.
-"""
-from argparse import ArgumentParser
-from enum import IntEnum
-import json
-import logging
-import sys
-import time
-from typing import Any, Callable, Optional, Tuple, Type, Union
-from urllib.parse import urljoin
-from urllib3 import PoolManager
-from urllib3.exceptions import MaxRetryError
-from urllib3.response import HTTPResponse
-
-
-class ExitCode(IntEnum):
-    """Possible exit codes of this script.
-
-    :cvar RECOVERY_DONE: backup was successfully restored.
-    :cvar RECOVERY_FAILED: recovery of the backup faced an issue.
-    :cvar API_NOT_OK: ``pg-backup-api`` status is not ``OK``.
-    :cvar HTTP_REQUEST_ERROR: an error has occurred during a request to the
-        ``pg-backup-api``.
-    :cvar HTTP_RESPONSE_MALFORMED: ``pg-backup-api`` returned a bogus response.
-    """
-
-    RECOVERY_DONE = 0
-    RECOVERY_FAILED = 1
-    API_NOT_OK = 2
-    HTTP_REQUEST_ERROR = 3
-    HTTP_RESPONSE_MALFORMED = 4
-
-
-class RetriesExceeded(Exception):
-    """Maximum number of retries exceeded."""
-
-
-def retry(exceptions: Union[Type[Exception], Tuple[Type[Exception], ...]]) \
-        -> Any:
-    """Retry an operation n times if expected *exceptions* are faced.
-
-    .. note::
-        Should be used as a decorator of a class' method as it expects the
-        first argument to be a class instance.
-
-        The class which method is going to be decorated should contain a couple
-        attributes:
-
-        * ``max_retries``: maximum retry attempts before failing;
-        * ``retry_wait``: how long to wait before retrying.
-
-    :param exceptions: exceptions that could trigger a retry attempt.
-
-    :raises:
-        :exc:`RetriesExceeded`: if the maximum number of attempts has been
-            exhausted.
-    """
-    def decorator(func: Callable[..., Any]) -> Any:
-        def inner_func(instance: object, *args: Any, **kwargs: Any) -> Any:
-            times: int = getattr(instance, "max_retries")
-            retry_wait: int = getattr(instance, "retry_wait")
-            method_name = f"{instance.__class__.__name__}.{func.__name__}"
-
-            attempt = 1
-
-            while attempt <= times:
-                try:
-                    return func(instance, *args, **kwargs)
-                except exceptions as exc:
-                    logging.warning("Attempt %d of %d on method %s failed "
-                                    "with %r.",
-                                    attempt, times, method_name, exc)
-                    attempt += 1
-
-                time.sleep(retry_wait)
-
-            raise RetriesExceeded("Maximum number of retries exceeded for "
-                                  f"method {method_name}.")
-        return inner_func
-    return decorator
-
-
-class BarmanRecover:
-    """Facilities for performing a remote ``barman recover`` operation.
-
-    You should instantiate this class, which will take care of configuring the
-    operation accordingly. When you want to start the operation, you should
-    call :meth:`restore_backup`. At any point of interaction with this class,
-    you may face a :func:`sys.exit` call. Refer to :class:`ExitCode` for a view
-    on the possible exit codes.
-
-    :ivar api_url: base URL to reach the ``pg-backup-api``.
-    :ivar cert_file: certificate to authenticate against the
-        ``pg-backup-api``, if required.
-    :ivar key_file: certificate key to authenticate against the
-        ``pg-backup-api``, if required.
-    :ivar barman_server: name of the Barman server which backup is to be
-        restored.
-    :ivar backup_id: ID of the backup from the Barman server.
-    :ivar ssh_command: SSH command to connect from the Barman host to the
-        local host.
-    :ivar data_directory: path to the Postgres data directory where to
-        restore the backup at.
-    :ivar loop_wait: how long to wait before checking again the status of the
-        recovery process. Higher values are useful for backups that are
-        expected to take long to restore.
-    :ivar retry_wait: how long to wait before retrying a failed request to the
-        ``pg-backup-api``.
-    :ivar max_retries: maximum number of retries when ``pg-backup-api`` returns
-        malformed responses.
-    :ivar http: a HTTP pool manager for performing web requests.
-    """
-
-    def __init__(self, api_url: str, barman_server: str, backup_id: str,
-                 ssh_command: str, data_directory: str, loop_wait: int,
-                 retry_wait: int, max_retries: int,
-                 cert_file: Optional[str] = None,
-                 key_file: Optional[str] = None) -> None:
-        """Create a new instance of :class:`BarmanRecover`.
-
-        Make sure the ``pg-backup-api`` is reachable and running fine.
-
-        :param api_url: base URL to reach the ``pg-backup-api``.
-        :param barman_server: name of the Barman server which backup is to be
-            restored.
-        :param backup_id: ID of the backup from the Barman server.
-        :param ssh_command: SSH command to connect from the Barman host to the
-            local host.
-        :param data_directory: path to the Postgres data directory where to
-            restore the backup at.
-        :param loop_wait: how long to wait before checking again the status of
-            the recovery process. Higher values are useful for backups that are
-            expected to take long to restore.
-        :param retry_wait: how long to wait before retrying a failed request to
-            the ``pg-backup-api``.
-        :param max_retries: maximum number of retries when ``pg-backup-api``
-            returns malformed responses.
-        :param cert_file: certificate to authenticate against the
-            ``pg-backup-api``, if required.
-        :param key_file: certificate key to authenticate against the
-            ``pg-backup-api``, if required.
-        """
-        self.api_url = api_url
-        self.cert_file = cert_file
-        self.key_file = key_file
-        self.barman_server = barman_server
-        self.backup_id = backup_id
-        self.ssh_command = ssh_command
-        self.data_directory = data_directory
-        self.loop_wait = loop_wait
-        self.retry_wait = retry_wait
-        self.max_retries = max_retries
-        self.http = PoolManager(cert_file=cert_file, key_file=key_file)
-        self._ensure_api_ok()
-
-    def _build_full_url(self, url_path: str) -> str:
-        """Build the full URL by concatenating *url_path* with the base URL.
-
-        :param url_path: path to be accessed in the ``pg-backup-api``.
-
-        :returns: the full URL after concatenating.
-        """
-        return urljoin(self.api_url, url_path)
-
-    @staticmethod
-    def _deserialize_response(response: HTTPResponse) -> Any:
-        """Retrieve body from *response* as a deserialized JSON object.
-
-        :param response: response from which JSON body will be deserialized.
-
-        :returns: the deserialized JSON body.
-        """
-        return json.loads(response.data.decode("utf-8"))
-
-    @staticmethod
-    def _serialize_request(body: Any) -> Any:
-        """Serialize a request body.
-
-        :param body: content of the request body to be serialized.
-
-        :returns: the serialized request body.
-        """
-        return json.dumps(body).encode("utf-8")
-
-    def _get_request(self, url_path: str) -> Any:
-        """Perform a ``GET`` request to *url_path*.
-
-        .. note::
-            If a :exc:`MaxRetryError` is faced while performing the request,
-            then exit with :attr:`ExitCode.HTTP_REQUEST_ERROR`
-
-        :param url_path: URL to perform the ``GET`` request against.
-
-        :returns: the deserialized response body.
-        """
-        response = None
-
-        try:
-            response = self.http.request("GET", self._build_full_url(url_path))
-        except MaxRetryError as exc:
-            logging.critical("An error occurred while performing an HTTP GET "
-                             "request: %r", exc)
-            sys.exit(ExitCode.HTTP_REQUEST_ERROR)
-
-        return self._deserialize_response(response)
-
-    def _post_request(self, url_path: str, body: Any) -> Any:
-        """Perform a ``POST`` request to *url_path* serializing *body* as JSON.
-
-        .. note::
-            If a :exc:`MaxRetryError` is faced while performing the request,
-            then exit with :attr:`ExitCode.HTTP_REQUEST_ERROR`
-
-        :param url_path: URL to perform the ``POST`` request against.
-        :param body: the body to be serialized as JSON and sent in the request.
-
-        :returns: the deserialized response body.
-        """
-        body = self._serialize_request(body)
-
-        response = None
-
-        try:
-            response = self.http.request("POST",
-                                         self._build_full_url(url_path),
-                                         body=body,
-                                         headers={
-                                             "Content-Type": "application/json"
-                                         })
-        except MaxRetryError as exc:
-            logging.critical("An error occurred while performing an HTTP POST "
-                             "request: %r", exc)
-            sys.exit(ExitCode.HTTP_REQUEST_ERROR)
-
-        return self._deserialize_response(response)
-
-    def _ensure_api_ok(self) -> None:
-        """Ensure ``pg-backup-api`` is reachable and ``OK``.
-
-        .. note::
-            If ``pg-backup-api`` status is not ``OK``, then exit with
-            :attr:`ExitCode.API_NOT_OK`.
-        """
-        response = self._get_request("status")
-
-        if response != "OK":
-            logging.critical("pg-backup-api is not working: %s", response)
-            sys.exit(ExitCode.API_NOT_OK)
-
-    @retry(KeyError)
-    def _create_recovery_operation(self) -> str:
-        """Create a recovery operation on the ``pg-backup-api``.
-
-        :returns: the ID of the recovery operation that has been created.
-        """
-        response = self._post_request(
-            f"servers/{self.barman_server}/operations",
-            {
-                "type": "recovery",
-                "backup_id": self.backup_id,
-                "remote_ssh_command": self.ssh_command,
-                "destination_directory": self.data_directory,
-            },
-        )
-
-        return response["operation_id"]
-
-    @retry(KeyError)
-    def _get_recovery_operation_status(self, operation_id: str) -> str:
-        """Get status of the recovery operation *operation_id*.
-
-        :param operation_id: ID of the recovery operation to be checked.
-
-        :returns: the status of the recovery operation.
-        """
-        response = self._get_request(
-            f"servers/{self.barman_server}/operations/{operation_id}",
-        )
-
-        return response["status"]
-
-    def restore_backup(self) -> bool:
-        """Restore the configured Barman backup through ``pg-backup-api``.
-
-        .. note::
-            If recovery API request returns a malformed response, then exit with
-            :attr:`ExitCode.HTTP_RESPONSE_MALFORMED`.
-
-        :returns: ``True`` if it was successfully recovered, ``False``
-            otherwise.
-        """
-        operation_id = None
-
-        try:
-            operation_id = self._create_recovery_operation()
-        except RetriesExceeded:
-            logging.critical("Maximum number of retries exceeded, exiting.")
-            sys.exit(ExitCode.HTTP_RESPONSE_MALFORMED)
-
-        logging.info("Created the recovery operation with ID %s", operation_id)
-
-        status = None
-
-        while True:
-            try:
-                status = self._get_recovery_operation_status(operation_id)
-            except RetriesExceeded:
-                logging.critical("Maximum number of retries exceeded, "
-                                 "exiting.")
-                sys.exit(ExitCode.HTTP_RESPONSE_MALFORMED)
-
-            if status != "IN_PROGRESS":
-                break
-
-            logging.info("Recovery operation %s is still in progress",
-                         operation_id)
-            time.sleep(self.loop_wait)
-
-        return status == "DONE"
-
-
-def set_up_logging(log_file: Optional[str] = None) -> None:
-    """Set up logging to file, if *log_file* is given, otherwise to console.
-
-    :param log_file: file where to log messages, if any.
-    """
-    logging.basicConfig(filename=log_file, level=logging.INFO,
-                        format="%(asctime)s %(levelname)s: %(message)s")
-
-
-def main() -> None:
-    """Entry point of this script.
-
-    Parse the command-line arguments and recover a Barman backup through
-    ``pg-backup-api`` to the local host.
-    """
-    parser = ArgumentParser(
-        epilog=(
-            "Wrapper script for ``pg-backup-api``. Communicate with the API "
-            "running  at ``--api-url`` to restore a ``--backup-id`` Barman "
-            "backup of the server ``--barman-server``."
-        ),
-    )
-    parser.add_argument(
-        "--api-url",
-        type=str,
-        required=True,
-        help="URL to reach the ``pg-backup-api``, e.g. "
-             "``http://localhost:7480``",
-        dest="api_url",
-    )
-    parser.add_argument(
-        "--cert-file",
-        type=str,
-        required=False,
-        help="Certificate to authenticate against the API, if required.",
-        dest="cert_file",
-    )
-    parser.add_argument(
-        "--key-file",
-        type=str,
-        required=False,
-        help="Certificate key to authenticate against the API, if required.",
-        dest="key_file",
-    )
-    parser.add_argument(
-        "--barman-server",
-        type=str,
-        required=True,
-        help="Name of the Barman server from which to restore the backup.",
-        dest="barman_server",
-    )
-    parser.add_argument(
-        "--backup-id",
-        type=str,
-        required=False,
-        default="latest",
-        help="ID of the Barman backup to be restored. You can use any value "
-             "supported by ``barman recover`` command "
-             "(default: ``%(default)s``)",
-        dest="backup_id",
-    )
-    parser.add_argument(
-        "--ssh-command",
-        type=str,
-        required=True,
-        help="Value to be passed as ``--remote-ssh-command`` to "
-             "``barman recover``.",
-        dest="ssh_command",
-    )
-    parser.add_argument(
-        "--data-directory",
-        "--datadir",
-        type=str,
-        required=True,
-        help="Destination path where to restore the barman backup in the "
-             "local host.",
-        dest="data_directory",
-    )
-    parser.add_argument(
-        "--log-file",
-        type=str,
-        required=False,
-        help="File where to log messages produced by this script, if any.",
-        dest="log_file",
-    )
-    parser.add_argument(
-        "--loop-wait",
-        type=int,
-        required=False,
-        default=10,
-        help="How long to wait before checking again the status of the "
-             "recovery process, in seconds. Use higher values if your "
-             "recovery is expected to take long (default: ``%(default)s``)",
-        dest="loop_wait",
-    )
-    parser.add_argument(
-        "--retry-wait",
-        type=int,
-        required=False,
-        default=2,
-        help="How long to wait before retrying a failed ``pg-backup-api`` "
-             "request (default: ``%(default)s``)",
-        dest="retry_wait",
-    )
-    parser.add_argument(
-        "--max-retries",
-        type=int,
-        required=False,
-        default=5,
-        help="Maximum number of retries when receiving malformed responses "
-             "from the ``pg-backup-api`` (default: ``%(default)s``)",
-        dest="max_retries",
-    )
-    args, _ = parser.parse_known_args()
-
-    set_up_logging(args.log_file)
-
-    barman_recover = BarmanRecover(args.api_url, args.barman_server,
-                                   args.backup_id, args.ssh_command,
-                                   args.data_directory, args.loop_wait,
-                                   args.retry_wait, args.max_retries,
-                                   args.cert_file, args.key_file)
-
-    successful = barman_recover.restore_backup()
-
-    if successful:
-        logging.info("Recovery operation finished successfully.")
-        sys.exit(ExitCode.RECOVERY_DONE)
-    else:
-        logging.critical("Recovery operation failed.")
-        sys.exit(ExitCode.RECOVERY_FAILED)
-
-
-if __name__ == "__main__":
-    main()

patroni/tags.py

@@ -3,13 +3,16 @@ import abc
 
 from typing import Any, Dict, Optional
 
-from patroni.utils import parse_int
+from patroni.utils import parse_int, parse_bool
 
 
 class Tags(abc.ABC):
     """An abstract class that encapsulates all the ``tags`` logic.
 
     Child classes that want to use provided facilities must implement ``tags`` abstract property.
+
+    .. note::
+        Due to backward-compatibility reasons, old tags may have a less strict type conversion than new ones.
     """
 
     @staticmethod
@@ -20,7 +23,7 @@ class Tags(abc.ABC):
 
         .. note::
             A custom tag is any tag added to the configuration ``tags`` section that is not one of ``clonefrom``,
-            ``nofailover``, ``noloadbalance`` or ``nosync``.
+            ``nofailover``, ``noloadbalance``,``nosync`` or ``nostream``.
 
             For most of the Patroni predefined tags, the returning object will only contain them if they are enabled as
             they all are boolean values that default to disabled.
@@ -31,7 +34,7 @@ class Tags(abc.ABC):
             tag value.
         """
         return {tag: value for tag, value in tags.items()
-                if any((tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync'),
+                if any((tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync', 'nostream'),
                         value,
                         tag == 'nofailover' and 'failover_priority' in tags))}
 
@@ -89,3 +92,8 @@ class Tags(abc.ABC):
     def replicatefrom(self) -> Optional[str]:
         """Value of ``replicatefrom`` tag, if any."""
         return self.tags.get('replicatefrom')
+
+    @property
+    def nostream(self) -> bool:
+        """``True`` if ``nostream`` is ``True``, else ``False``."""
+        return parse_bool(self.tags.get('nostream')) or False

patroni/utils.py

@@ -716,7 +716,7 @@ class Retry(object):
         return self._cur_stoptime or 0
 
     def ensure_deadline(self, timeout: float, raise_ex: Optional[Exception] = None) -> bool:
-        """Calculates, sets, and checks the remaining deadline time.
+        """Calculates and checks the remaining deadline time.
 
         :param timeout: if the *deadline* is smaller than the provided *timeout* value raise *raise_ex* exception.
         :param raise_ex: the exception object that will be raised if the *deadline* is smaller than provided *timeout*.
@@ -727,8 +727,7 @@ class Retry(object):
         :raises:
             :class:`Exception`: *raise_ex* if calculated deadline is smaller than provided *timeout*.
         """
-        self.deadline = self.stoptime - time.time()
-        if self.deadline < timeout:
+        if self.stoptime - time.time() < timeout:
             if raise_ex:
                 raise raise_ex
             return False

patroni/validator.py

@@ -11,8 +11,7 @@ import socket
 
 from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING
 
-from .collections import CaseInsensitiveSet
-
+from .collections import CaseInsensitiveSet, EMPTY_DICT
 from .dcs import dcs_modules
 from .exceptions import ConfigParseError
 from .utils import parse_int, split_host_port, data_directory_is_empty, get_major_version
@@ -245,7 +244,7 @@ def get_bin_name(bin_name: str) -> str:
     """
     if TYPE_CHECKING:  # pragma: no cover
         assert isinstance(schema.data, dict)
-    return (schema.data.get('postgresql', {}).get('bin_name', {}) or {}).get(bin_name, bin_name)
+    return (schema.data.get('postgresql', {}).get('bin_name', {}) or EMPTY_DICT).get(bin_name, bin_name)
 
 
 def validate_data_dir(data_dir: str) -> bool:
@@ -1172,6 +1171,7 @@ schema = Schema({
         Optional("clonefrom"): bool,
         Optional("noloadbalance"): bool,
         Optional("replicatefrom"): str,
-        Optional("nosync"): bool
+        Optional("nosync"): bool,
+        Optional("nostream"): bool
     }
 })

patroni/version.py

@@ -2,4 +2,4 @@
 
 :var __version__: the current Patroni version.
 """
-__version__ = '3.2.2'
+__version__ = '3.3.0'

postgres0.yml

@@ -136,3 +136,4 @@ tags:
     noloadbalance: false
     clonefrom: false
     nosync: false
+    nostream: false

setup.py

@@ -55,7 +55,7 @@ CONSOLE_SCRIPTS = ['patroni = patroni.__main__:main',
                    'patroni_raft_controller = patroni.raft_controller:main',
                    "patroni_wale_restore = patroni.scripts.wale_restore:main",
                    "patroni_aws = patroni.scripts.aws:main",
-                   "patroni_barman_recover = patroni.scripts.barman_recover:main"]
+                   "patroni_barman = patroni.scripts.barman.cli:main"]
 
 
 class _Command(Command):

tests/test_barman.py

@@ -0,0 +1,765 @@
+import logging
+import mock
+from mock import MagicMock, Mock, patch
+import unittest
+from urllib3.exceptions import MaxRetryError
+
+from patroni.scripts.barman.cli import main
+from patroni.scripts.barman.config_switch import (ExitCode as BarmanConfigSwitchExitCode, _should_skip_switch,
+                                                  _switch_config, run_barman_config_switch)
+from patroni.scripts.barman.recover import ExitCode as BarmanRecoverExitCode, _restore_backup, run_barman_recover
+from patroni.scripts.barman.utils import ApiNotOk, OperationStatus, PgBackupApi, RetriesExceeded, set_up_logging
+
+
+API_URL = "http://localhost:7480"
+BARMAN_SERVER = "my_server"
+BARMAN_MODEL = "my_model"
+BACKUP_ID = "backup_id"
+SSH_COMMAND = "ssh postgres@localhost"
+DATA_DIRECTORY = "/path/to/pgdata"
+LOOP_WAIT = 10
+RETRY_WAIT = 2
+MAX_RETRIES = 5
+
+
+# stuff from patroni.scripts.barman.utils
+
+@patch("logging.basicConfig")
+def test_set_up_logging(mock_log_config):
+    log_file = "/path/to/some/file.log"
+    set_up_logging(log_file)
+    mock_log_config.assert_called_once_with(filename=log_file, level=logging.INFO,
+                                            format="%(asctime)s %(levelname)s: %(message)s")
+
+
+class TestPgBackupApi(unittest.TestCase):
+
+    @patch.object(PgBackupApi, "_ensure_api_ok", Mock())
+    @patch("patroni.scripts.barman.utils.PoolManager", MagicMock())
+    def setUp(self):
+        self.api = PgBackupApi(API_URL, None, None, RETRY_WAIT, MAX_RETRIES)
+        # Reset the mock as the same instance is used across tests
+        self.api._http.request.reset_mock()
+        self.api._http.request.side_effect = None
+
+    def test__build_full_url(self):
+        self.assertEqual(self.api._build_full_url("/some/path"), f"{API_URL}/some/path")
+
+    @patch("json.loads")
+    def test__deserialize_response(self, mock_json_loads):
+        mock_response = MagicMock()
+        self.assertIsNotNone(self.api._deserialize_response(mock_response))
+        mock_json_loads.assert_called_once_with(mock_response.data.decode("utf-8"))
+
+    @patch("json.dumps")
+    def test__serialize_request(self, mock_json_dumps):
+        body = "some_body"
+        ret = self.api._serialize_request(body)
+        self.assertIsNotNone(ret)
+        mock_json_dumps.assert_called_once_with(body)
+        mock_json_dumps.return_value.encode.assert_called_once_with("utf-8")
+
+    @patch.object(PgBackupApi, "_deserialize_response", Mock(return_value="test"))
+    def test__get_request(self):
+        mock_request = self.api._http.request
+
+        # with no error
+        self.assertEqual(self.api._get_request("/some/path"), "test")
+        mock_request.assert_called_once_with("GET", f"{API_URL}/some/path")
+
+        # with MaxRetryError
+        http_error = MaxRetryError(self.api._http, f"{API_URL}/some/path")
+        mock_request.side_effect = http_error
+
+        with self.assertRaises(RetriesExceeded) as exc:
+            self.assertIsNone(self.api._get_request("/some/path"))
+
+        self.assertEqual(
+            str(exc.exception),
+            "Failed to perform a GET request to http://localhost:7480/some/path"
+        )
+
+    @patch.object(PgBackupApi, "_deserialize_response", Mock(return_value="test"))
+    @patch.object(PgBackupApi, "_serialize_request")
+    def test__post_request(self, mock_serialize):
+        mock_request = self.api._http.request
+
+        # with no error
+        self.assertEqual(self.api._post_request("/some/path", "some body"), "test")
+        mock_serialize.assert_called_once_with("some body")
+        mock_request.assert_called_once_with("POST", f"{API_URL}/some/path", body=mock_serialize.return_value,
+                                             headers={"Content-Type": "application/json"})
+
+        # with HTTPError
+        http_error = MaxRetryError(self.api._http, f"{API_URL}/some/path")
+        mock_request.side_effect = http_error
+
+        with self.assertRaises(RetriesExceeded) as exc:
+            self.assertIsNone(self.api._post_request("/some/path", "some body"))
+
+        self.assertEqual(
+            str(exc.exception),
+            f"Failed to perform a POST request to http://localhost:7480/some/path with {mock_serialize.return_value}"
+        )
+
+    @patch.object(PgBackupApi, "_get_request")
+    def test__ensure_api_ok(self, mock_get_request):
+        # API ok
+        mock_get_request.return_value = "OK"
+        self.assertIsNone(self.api._ensure_api_ok())
+
+        # API not ok
+        mock_get_request.return_value = "random"
+
+        with self.assertRaises(ApiNotOk) as exc:
+            self.assertIsNone(self.api._ensure_api_ok())
+
+        self.assertEqual(
+            str(exc.exception),
+            "pg-backup-api is currently not up and running at http://localhost:7480: random",
+        )
+
+    @patch("patroni.scripts.barman.utils.OperationStatus")
+    @patch("logging.warning")
+    @patch("time.sleep")
+    @patch.object(PgBackupApi, "_get_request")
+    def test_get_operation_status(self, mock_get_request, mock_sleep, mock_logging, mock_op_status):
+        # well formed response
+        mock_get_request.return_value = {"status": "some status"}
+        mock_op_status.__getitem__.return_value = "SOME_STATUS"
+        self.assertEqual(self.api.get_operation_status(BARMAN_SERVER, "some_id"), "SOME_STATUS")
+        mock_get_request.assert_called_once_with(f"servers/{BARMAN_SERVER}/operations/some_id")
+        mock_sleep.assert_not_called()
+        mock_logging.assert_not_called()
+        mock_op_status.__getitem__.assert_called_once_with("some status")
+
+        # malformed response
+        mock_get_request.return_value = {"statuss": "some status"}
+
+        with self.assertRaises(RetriesExceeded) as exc:
+            self.api.get_operation_status(BARMAN_SERVER, "some_id")
+
+        self.assertEqual(str(exc.exception),
+                         "Maximum number of retries exceeded for method PgBackupApi.get_operation_status.")
+
+        self.assertEqual(mock_sleep.call_count, self.api.max_retries)
+        mock_sleep.assert_has_calls([mock.call(self.api.retry_wait)] * self.api.max_retries)
+
+        self.assertEqual(mock_logging.call_count, self.api.max_retries)
+        for i in range(mock_logging.call_count):
+            call_args = mock_logging.call_args_list[i][0]
+            self.assertEqual(len(call_args), 5)
+            self.assertEqual(call_args[0], "Attempt %d of %d on method %s failed with %r.")
+            self.assertEqual(call_args[1], i + 1)
+            self.assertEqual(call_args[2], self.api.max_retries)
+            self.assertEqual(call_args[3], "PgBackupApi.get_operation_status")
+            self.assertIsInstance(call_args[4], KeyError)
+            self.assertEqual(call_args[4].args, ('status',))
+
+    @patch("logging.warning")
+    @patch("time.sleep")
+    @patch.object(PgBackupApi, "_post_request")
+    def test_create_recovery_operation(self, mock_post_request, mock_sleep, mock_logging):
+        # well formed response
+        mock_post_request.return_value = {"operation_id": "some_id"}
+        self.assertEqual(
+            self.api.create_recovery_operation(BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY),
+            "some_id",
+        )
+        mock_sleep.assert_not_called()
+        mock_logging.assert_not_called()
+        mock_post_request.assert_called_once_with(
+            f"servers/{BARMAN_SERVER}/operations",
+            {
+                "type": "recovery",
+                "backup_id": BACKUP_ID,
+                "remote_ssh_command": SSH_COMMAND,
+                "destination_directory": DATA_DIRECTORY,
+            }
+        )
+
+        # malformed response
+        mock_post_request.return_value = {"operation_idd": "some_id"}
+
+        with self.assertRaises(RetriesExceeded) as exc:
+            self.api.create_recovery_operation(BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY)
+
+        self.assertEqual(str(exc.exception),
+                         "Maximum number of retries exceeded for method PgBackupApi.create_recovery_operation.")
+
+        self.assertEqual(mock_sleep.call_count, self.api.max_retries)
+
+        mock_sleep.assert_has_calls([mock.call(self.api.retry_wait)] * self.api.max_retries)
+
+        self.assertEqual(mock_logging.call_count, self.api.max_retries)
+        for i in range(mock_logging.call_count):
+            call_args = mock_logging.call_args_list[i][0]
+            self.assertEqual(len(call_args), 5)
+            self.assertEqual(call_args[0], "Attempt %d of %d on method %s failed with %r.")
+            self.assertEqual(call_args[1], i + 1)
+            self.assertEqual(call_args[2], self.api.max_retries)
+            self.assertEqual(call_args[3], "PgBackupApi.create_recovery_operation")
+            self.assertIsInstance(call_args[4], KeyError)
+            self.assertEqual(call_args[4].args, ('operation_id',))
+
+    @patch("logging.warning")
+    @patch("time.sleep")
+    @patch.object(PgBackupApi, "_post_request")
+    def test_create_config_switch_operation(self, mock_post_request, mock_sleep, mock_logging):
+        # well formed response -- sample 1
+        mock_post_request.return_value = {"operation_id": "some_id"}
+        self.assertEqual(
+            self.api.create_config_switch_operation(BARMAN_SERVER, BARMAN_MODEL, None),
+            "some_id",
+        )
+        mock_sleep.assert_not_called()
+        mock_logging.assert_not_called()
+        mock_post_request.assert_called_once_with(
+            f"servers/{BARMAN_SERVER}/operations",
+            {
+                "type": "config_switch",
+                "model_name": BARMAN_MODEL,
+            }
+        )
+
+        # well formed response -- sample 2
+        mock_post_request.reset_mock()
+
+        self.assertEqual(
+            self.api.create_config_switch_operation(BARMAN_SERVER, None, True),
+            "some_id",
+        )
+        mock_sleep.assert_not_called()
+        mock_logging.assert_not_called()
+        mock_post_request.assert_called_once_with(
+            f"servers/{BARMAN_SERVER}/operations",
+            {
+                "type": "config_switch",
+                "reset": True,
+            }
+        )
+
+        # malformed response
+        mock_post_request.return_value = {"operation_idd": "some_id"}
+
+        with self.assertRaises(RetriesExceeded) as exc:
+            self.api.create_config_switch_operation(BARMAN_SERVER, BARMAN_MODEL, None)
+
+        self.assertEqual(str(exc.exception),
+                         "Maximum number of retries exceeded for method PgBackupApi.create_config_switch_operation.")
+
+        self.assertEqual(mock_sleep.call_count, self.api.max_retries)
+
+        mock_sleep.assert_has_calls([mock.call(self.api.retry_wait)] * self.api.max_retries)
+
+        self.assertEqual(mock_logging.call_count, self.api.max_retries)
+        for i in range(mock_logging.call_count):
+            call_args = mock_logging.call_args_list[i][0]
+            self.assertEqual(len(call_args), 5)
+            self.assertEqual(call_args[0], "Attempt %d of %d on method %s failed with %r.")
+            self.assertEqual(call_args[1], i + 1)
+            self.assertEqual(call_args[2], self.api.max_retries)
+            self.assertEqual(call_args[3], "PgBackupApi.create_config_switch_operation")
+            self.assertIsInstance(call_args[4], KeyError)
+            self.assertEqual(call_args[4].args, ('operation_id',))
+
+
+# stuff from patroni.scripts.barman.recover
+
+
+class TestBarmanRecover(unittest.TestCase):
+
+    def setUp(self):
+        self.api = MagicMock()
+        # Reset the mock as the same instance is used across tests
+        self.api._http.request.reset_mock()
+        self.api._http.request.side_effect = None
+
+    @patch("time.sleep")
+    @patch("logging.info")
+    @patch("logging.error")
+    def test__restore_backup(self, mock_log_error, mock_log_info, mock_sleep):
+        mock_create_op = self.api.create_recovery_operation
+        mock_get_status = self.api.get_operation_status
+
+        # successful fast restore
+        mock_create_op.return_value = "some_id"
+        mock_get_status.return_value = OperationStatus.DONE
+
+        self.assertEqual(
+            _restore_backup(self.api, BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY, LOOP_WAIT),
+            BarmanRecoverExitCode.RECOVERY_DONE,
+        )
+
+        mock_create_op.assert_called_once_with(BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY)
+        mock_get_status.assert_called_once_with(BARMAN_SERVER, "some_id")
+        mock_log_info.assert_has_calls([
+            mock.call("Created the recovery operation with ID %s", "some_id"),
+            mock.call("Recovery operation finished successfully."),
+        ])
+        mock_log_error.assert_not_called()
+        mock_sleep.assert_not_called()
+
+        # successful slow restore
+        mock_create_op.reset_mock()
+        mock_get_status.reset_mock()
+        mock_log_info.reset_mock()
+        mock_get_status.side_effect = [OperationStatus.IN_PROGRESS] * 20 + [OperationStatus.DONE]
+
+        self.assertEqual(
+            _restore_backup(self.api, BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY, LOOP_WAIT),
+            BarmanRecoverExitCode.RECOVERY_DONE,
+        )
+
+        mock_create_op.assert_called_once()
+
+        self.assertEqual(mock_get_status.call_count, 21)
+        mock_get_status.assert_has_calls([mock.call(BARMAN_SERVER, "some_id")] * 21)
+
+        self.assertEqual(mock_log_info.call_count, 22)
+        mock_log_info.assert_has_calls([mock.call("Created the recovery operation with ID %s", "some_id")]
+                                       + [mock.call("Recovery operation %s is still in progress", "some_id")] * 20
+                                       + [mock.call("Recovery operation finished successfully.")])
+
+        mock_log_error.assert_not_called()
+
+        self.assertEqual(mock_sleep.call_count, 20)
+        mock_sleep.assert_has_calls([mock.call(LOOP_WAIT)] * 20)
+
+        # failed fast restore
+        mock_create_op.reset_mock()
+        mock_get_status.reset_mock()
+        mock_log_info.reset_mock()
+        mock_sleep.reset_mock()
+        mock_get_status.side_effect = None
+        mock_get_status.return_value = OperationStatus.FAILED
+
+        self.assertEqual(
+            _restore_backup(self.api, BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY, LOOP_WAIT),
+            BarmanRecoverExitCode.RECOVERY_FAILED,
+        )
+
+        mock_create_op.assert_called_once()
+        mock_get_status.assert_called_once_with(BARMAN_SERVER, "some_id")
+        mock_log_info.assert_has_calls([
+            mock.call("Created the recovery operation with ID %s", "some_id"),
+        ])
+        mock_log_error.assert_has_calls([
+            mock.call("Recovery operation failed."),
+        ])
+        mock_sleep.assert_not_called()
+
+        # failed slow restore
+        mock_create_op.reset_mock()
+        mock_get_status.reset_mock()
+        mock_log_info.reset_mock()
+        mock_log_error.reset_mock()
+        mock_sleep.reset_mock()
+        mock_get_status.side_effect = [OperationStatus.IN_PROGRESS] * 20 + [OperationStatus.FAILED]
+
+        self.assertEqual(
+            _restore_backup(self.api, BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY, LOOP_WAIT),
+            BarmanRecoverExitCode.RECOVERY_FAILED,
+        )
+
+        mock_create_op.assert_called_once()
+
+        self.assertEqual(mock_get_status.call_count, 21)
+        mock_get_status.assert_has_calls([mock.call(BARMAN_SERVER, "some_id")] * 21)
+
+        self.assertEqual(mock_log_info.call_count, 21)
+        mock_log_info.assert_has_calls([mock.call("Created the recovery operation with ID %s", "some_id")]
+                                       + [mock.call("Recovery operation %s is still in progress", "some_id")] * 20)
+
+        mock_log_error.assert_has_calls([
+            mock.call("Recovery operation failed."),
+        ])
+
+        self.assertEqual(mock_sleep.call_count, 20)
+        mock_sleep.assert_has_calls([mock.call(LOOP_WAIT)] * 20)
+
+        # create retries exceeded
+        mock_log_info.reset_mock()
+        mock_log_error.reset_mock()
+        mock_sleep.reset_mock()
+        mock_create_op.side_effect = RetriesExceeded()
+        mock_get_status.side_effect = None
+
+        self.assertEqual(
+            _restore_backup(self.api, BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY, LOOP_WAIT),
+            BarmanRecoverExitCode.HTTP_ERROR,
+        )
+
+        mock_log_info.assert_not_called()
+        mock_log_error.assert_called_once_with("An issue was faced while trying to create a recovery operation: %r",
+                                               mock_create_op.side_effect)
+        mock_sleep.assert_not_called()
+
+        # get status retries exceeded
+        mock_create_op.reset_mock()
+        mock_create_op.side_effect = None
+        mock_log_error.reset_mock()
+        mock_log_info.reset_mock()
+        mock_get_status.side_effect = RetriesExceeded
+
+        self.assertEqual(
+            _restore_backup(self.api, BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY, LOOP_WAIT),
+            BarmanRecoverExitCode.HTTP_ERROR,
+        )
+
+        mock_log_info.assert_called_once_with("Created the recovery operation with ID %s", "some_id")
+        mock_log_error.assert_called_once_with("Maximum number of retries exceeded, exiting.")
+        mock_sleep.assert_not_called()
+
+
+class TestBarmanRecoverCli(unittest.TestCase):
+
+    @patch("patroni.scripts.barman.recover._restore_backup")
+    def test_run_barman_recover(self, mock_rb):
+        api = MagicMock()
+        args = MagicMock()
+
+        # successful execution
+        mock_rb.return_value = BarmanRecoverExitCode.RECOVERY_DONE
+
+        self.assertEqual(
+            run_barman_recover(api, args),
+            BarmanRecoverExitCode.RECOVERY_DONE,
+        )
+
+        mock_rb.assert_called_once_with(api, args.barman_server, args.backup_id,
+                                        args.ssh_command, args.data_directory,
+                                        args.loop_wait)
+
+        # failed execution
+        mock_rb.reset_mock()
+
+        mock_rb.return_value = BarmanRecoverExitCode.RECOVERY_FAILED
+
+        self.assertEqual(
+            run_barman_recover(api, args),
+            BarmanRecoverExitCode.RECOVERY_FAILED,
+        )
+
+        mock_rb.assert_called_once_with(api, args.barman_server, args.backup_id,
+                                        args.ssh_command, args.data_directory,
+                                        args.loop_wait)
+
+
+# stuff from patroni.scripts.barman.config_switch
+
+
+class TestBarmanConfigSwitch(unittest.TestCase):
+
+    def setUp(self):
+        self.api = MagicMock()
+        # Reset the mock as the same instance is used across tests
+        self.api._http.request.reset_mock()
+        self.api._http.request.side_effect = None
+
+    @patch("time.sleep")
+    @patch("logging.info")
+    @patch("logging.error")
+    def test__switch_config(self, mock_log_error, mock_log_info, mock_sleep):
+        mock_create_op = self.api.create_config_switch_operation
+        mock_get_status = self.api.get_operation_status
+
+        # successful fast config-switch
+        mock_create_op.return_value = "some_id"
+        mock_get_status.return_value = OperationStatus.DONE
+
+        self.assertEqual(
+            _switch_config(self.api, BARMAN_SERVER, BARMAN_MODEL, None),
+            BarmanConfigSwitchExitCode.CONFIG_SWITCH_DONE,
+        )
+
+        mock_create_op.assert_called_once_with(BARMAN_SERVER, BARMAN_MODEL, None)
+        mock_get_status.assert_called_once_with(BARMAN_SERVER, "some_id")
+        mock_log_info.assert_has_calls([
+            mock.call("Created the config switch operation with ID %s", "some_id"),
+            mock.call("Config switch operation finished successfully."),
+        ])
+        mock_log_error.assert_not_called()
+        mock_sleep.assert_not_called()
+
+        # successful slow config-switch
+        mock_create_op.reset_mock()
+        mock_get_status.reset_mock()
+        mock_log_info.reset_mock()
+        mock_get_status.side_effect = [OperationStatus.IN_PROGRESS] * 20 + [OperationStatus.DONE]
+
+        self.assertEqual(
+            _switch_config(self.api, BARMAN_SERVER, BARMAN_MODEL, None),
+            BarmanConfigSwitchExitCode.CONFIG_SWITCH_DONE,
+        )
+
+        mock_create_op.assert_called_once_with(BARMAN_SERVER, BARMAN_MODEL, None)
+
+        self.assertEqual(mock_get_status.call_count, 21)
+        mock_get_status.assert_has_calls([mock.call(BARMAN_SERVER, "some_id")] * 21)
+
+        self.assertEqual(mock_log_info.call_count, 22)
+        mock_log_info.assert_has_calls([mock.call("Created the config switch operation with ID %s", "some_id")]
+                                       + [mock.call("Config switch operation %s is still in progress", "some_id")] * 20
+                                       + [mock.call("Config switch operation finished successfully.")])
+
+        mock_log_error.assert_not_called()
+
+        self.assertEqual(mock_sleep.call_count, 20)
+        mock_sleep.assert_has_calls([mock.call(5)] * 20)
+
+        # failed fast config-switch
+        mock_create_op.reset_mock()
+        mock_get_status.reset_mock()
+        mock_log_info.reset_mock()
+        mock_sleep.reset_mock()
+        mock_get_status.side_effect = None
+        mock_get_status.return_value = OperationStatus.FAILED
+
+        self.assertEqual(
+            _switch_config(self.api, BARMAN_SERVER, BARMAN_MODEL, None),
+            BarmanConfigSwitchExitCode.CONFIG_SWITCH_FAILED,
+        )
+
+        mock_create_op.assert_called_once()
+        mock_get_status.assert_called_once_with(BARMAN_SERVER, "some_id")
+        mock_log_info.assert_called_once_with("Created the config switch operation with ID %s", "some_id")
+        mock_log_error.assert_called_once_with("Config switch operation failed.")
+        mock_sleep.assert_not_called()
+
+        # failed slow config-switch
+        mock_create_op.reset_mock()
+        mock_get_status.reset_mock()
+        mock_log_info.reset_mock()
+        mock_log_error.reset_mock()
+        mock_sleep.reset_mock()
+        mock_get_status.side_effect = [OperationStatus.IN_PROGRESS] * 20 + [OperationStatus.FAILED]
+
+        self.assertEqual(
+            _switch_config(self.api, BARMAN_SERVER, BARMAN_MODEL, None),
+            BarmanConfigSwitchExitCode.CONFIG_SWITCH_FAILED,
+        )
+
+        mock_create_op.assert_called_once()
+
+        self.assertEqual(mock_get_status.call_count, 21)
+        mock_get_status.assert_has_calls([mock.call(BARMAN_SERVER, "some_id")] * 21)
+
+        self.assertEqual(mock_log_info.call_count, 21)
+        mock_log_info.assert_has_calls([mock.call("Created the config switch operation with ID %s", "some_id")]
+                                       + [mock.call("Config switch operation %s is still in progress", "some_id")] * 20)
+
+        mock_log_error.assert_called_once_with("Config switch operation failed.")
+
+        self.assertEqual(mock_sleep.call_count, 20)
+        mock_sleep.assert_has_calls([mock.call(5)] * 20)
+
+        # create retries exceeded
+        mock_log_info.reset_mock()
+        mock_log_error.reset_mock()
+        mock_sleep.reset_mock()
+        mock_create_op.side_effect = RetriesExceeded()
+        mock_get_status.side_effect = None
+
+        self.assertEqual(
+            _switch_config(self.api, BARMAN_SERVER, BARMAN_MODEL, None),
+            BarmanConfigSwitchExitCode.HTTP_ERROR,
+        )
+
+        mock_log_info.assert_not_called()
+        mock_log_error.assert_called_once_with("An issue was faced while trying to create a config switch operation: "
+                                               "%r",
+                                               mock_create_op.side_effect)
+        mock_sleep.assert_not_called()
+
+        # get status retries exceeded
+        mock_create_op.reset_mock()
+        mock_create_op.side_effect = None
+        mock_log_error.reset_mock()
+        mock_get_status.side_effect = RetriesExceeded
+
+        self.assertEqual(
+            _switch_config(self.api, BARMAN_SERVER, BARMAN_MODEL, None),
+            BarmanConfigSwitchExitCode.HTTP_ERROR,
+        )
+
+        mock_log_info.assert_called_once_with("Created the config switch operation with ID %s", "some_id")
+        mock_log_error.assert_called_once_with("Maximum number of retries exceeded, exiting.")
+        mock_sleep.assert_not_called()
+
+
+class TestBarmanConfigSwitchCli(unittest.TestCase):
+
+    def test__should_skip_switch(self):
+        args = MagicMock()
+
+        for role, switch_when, expected in [
+            ("master", "promoted", False),
+            ("master", "demoted", True),
+            ("master", "always", False),
+
+            ("primary", "promoted", False),
+            ("primary", "demoted", True),
+            ("primary", "always", False),
+
+            ("promoted", "promoted", False),
+            ("promoted", "demoted", True),
+            ("promoted", "always", False),
+
+            ("standby_leader", "promoted", True),
+            ("standby_leader", "demoted", True),
+            ("standby_leader", "always", False),
+
+            ("replica", "promoted", True),
+            ("replica", "demoted", False),
+            ("replica", "always", False),
+
+            ("demoted", "promoted", True),
+            ("demoted", "demoted", False),
+            ("demoted", "always", False),
+        ]:
+            args.role = role
+            args.switch_when = switch_when
+            self.assertEqual(_should_skip_switch(args), expected)
+
+    @patch("patroni.scripts.barman.config_switch._should_skip_switch")
+    @patch("patroni.scripts.barman.config_switch._switch_config")
+    @patch("logging.error")
+    @patch("logging.info")
+    def test_run_barman_config_switch(self, mock_log_info, mock_log_error, mock_sc, mock_skip):
+        api = MagicMock()
+        args = MagicMock()
+        args.reset = None
+
+        # successful execution
+        mock_skip.return_value = False
+        mock_sc.return_value = BarmanConfigSwitchExitCode.CONFIG_SWITCH_DONE
+
+        self.assertEqual(
+            run_barman_config_switch(api, args),
+            BarmanConfigSwitchExitCode.CONFIG_SWITCH_DONE,
+        )
+
+        mock_sc.assert_called_once_with(api, args.barman_server, args.barman_model,
+                                        args.reset)
+
+        # failed execution
+        mock_sc.reset_mock()
+
+        mock_sc.return_value = BarmanConfigSwitchExitCode.CONFIG_SWITCH_FAILED
+
+        self.assertEqual(
+            run_barman_config_switch(api, args),
+            BarmanConfigSwitchExitCode.CONFIG_SWITCH_FAILED,
+        )
+
+        mock_sc.assert_called_once_with(api, args.barman_server, args.barman_model,
+                                        args.reset)
+
+        # skipped execution
+        mock_sc.reset_mock()
+        mock_skip.return_value = True
+
+        self.assertEqual(
+            run_barman_config_switch(api, args),
+            BarmanConfigSwitchExitCode.CONFIG_SWITCH_SKIPPED
+        )
+
+        mock_sc.assert_not_called()
+        mock_log_info.assert_called_once_with("Config switch operation was skipped (role=%s, "
+                                              "switch_when=%s).", args.role, args.switch_when)
+        mock_log_error.assert_not_called()
+
+        # invalid args -- sample 1
+        mock_skip.return_value = False
+        args = MagicMock()
+        args.barman_server = BARMAN_SERVER
+        args.barman_model = BARMAN_MODEL
+        args.reset = True
+
+        self.assertEqual(
+            run_barman_config_switch(api, args),
+            BarmanConfigSwitchExitCode.INVALID_ARGS,
+        )
+
+        mock_log_error.assert_called_once_with("One, and only one among 'barman_model' ('%s') and 'reset' "
+                                               "('%s') should be given", BARMAN_MODEL, True)
+        api.assert_not_called()
+
+        # invalid args -- sample 2
+        args = MagicMock()
+        args.barman_server = BARMAN_SERVER
+        args.barman_model = None
+        args.reset = None
+
+        mock_log_error.reset_mock()
+        api.reset_mock()
+
+        self.assertEqual(
+            run_barman_config_switch(api, args),
+            BarmanConfigSwitchExitCode.INVALID_ARGS,
+        )
+
+        mock_log_error.assert_called_once_with("One, and only one among 'barman_model' ('%s') and 'reset' "
+                                               "('%s') should be given", None, None)
+        api.assert_not_called()
+
+
+# stuff from patroni.scripts.barman.cli
+
+
+class TestMain(unittest.TestCase):
+
+    @patch("patroni.scripts.barman.cli.PgBackupApi")
+    @patch("patroni.scripts.barman.cli.set_up_logging")
+    @patch("patroni.scripts.barman.cli.ArgumentParser")
+    def test_main(self, mock_arg_parse, mock_set_up_log, mock_api):
+        # sub-command specified
+        args = MagicMock()
+        args.func.return_value = 0
+        mock_arg_parse.return_value.parse_known_args.return_value = (args, None)
+
+        with self.assertRaises(SystemExit) as exc:
+            main()
+
+        mock_arg_parse.assert_called_once()
+        mock_set_up_log.assert_called_once_with(args.log_file)
+        mock_api.assert_called_once_with(args.api_url, args.cert_file,
+                                         args.key_file, args.retry_wait,
+                                         args.max_retries)
+        mock_arg_parse.return_value.print_help.assert_not_called()
+        args.func.assert_called_once_with(mock_api.return_value, args)
+        self.assertEqual(exc.exception.code, 0)
+
+        # Issue in the API
+        mock_arg_parse.reset_mock()
+        mock_set_up_log.reset_mock()
+        mock_api.reset_mock()
+        mock_api.side_effect = ApiNotOk()
+
+        with self.assertRaises(SystemExit) as exc:
+            main()
+
+        mock_arg_parse.assert_called_once()
+        mock_set_up_log.assert_called_once_with(args.log_file)
+        mock_api.assert_called_once_with(args.api_url, args.cert_file,
+                                         args.key_file, args.retry_wait,
+                                         args.max_retries)
+        mock_arg_parse.return_value.print_help.assert_not_called()
+        self.assertEqual(exc.exception.code, -2)
+
+        # sub-command not specified
+        mock_arg_parse.reset_mock()
+        mock_set_up_log.reset_mock()
+        mock_api.reset_mock()
+        delattr(args, "func")
+        mock_api.side_effect = None
+
+        with self.assertRaises(SystemExit) as exc:
+            main()
+
+        mock_arg_parse.assert_called_once()
+        mock_set_up_log.assert_called_once_with(args.log_file)
+        mock_api.assert_not_called()
+        mock_arg_parse.return_value.print_help.assert_called_once_with()
+        self.assertEqual(exc.exception.code, -1)

tests/test_barman_recover.py

@@ -1,366 +0,0 @@
-import logging
-import mock
-from mock import MagicMock, Mock, patch
-import unittest
-from urllib3.exceptions import MaxRetryError
-
-from patroni.scripts.barman_recover import BarmanRecover, ExitCode, RetriesExceeded, main, set_up_logging
-
-
-API_URL = "http://localhost:7480"
-BARMAN_SERVER = "my_server"
-BACKUP_ID = "backup_id"
-SSH_COMMAND = "ssh postgres@localhost"
-DATA_DIRECTORY = "/path/to/pgdata"
-LOOP_WAIT = 10
-RETRY_WAIT = 2
-MAX_RETRIES = 5
-
-
-class TestBarmanRecover(unittest.TestCase):
-
-    @patch.object(BarmanRecover, "_ensure_api_ok", Mock())
-    @patch("patroni.scripts.barman_recover.PoolManager", MagicMock())
-    def setUp(self):
-        self.br = BarmanRecover(API_URL, BARMAN_SERVER, BACKUP_ID, SSH_COMMAND, DATA_DIRECTORY, LOOP_WAIT, RETRY_WAIT,
-                                MAX_RETRIES)
-        # Reset the mock as the same instance is used across tests
-        self.br.http.request.reset_mock()
-        self.br.http.request.side_effect = None
-
-    def test__build_full_url(self):
-        self.assertEqual(self.br._build_full_url("/some/path"), f"{API_URL}/some/path")
-
-    @patch("json.loads")
-    def test__deserialize_response(self, mock_json_loads):
-        mock_response = MagicMock()
-        self.assertIsNotNone(self.br._deserialize_response(mock_response))
-        mock_json_loads.assert_called_once_with(mock_response.data.decode("utf-8"))
-
-    @patch("json.dumps")
-    def test__serialize_request(self, mock_json_dumps):
-        body = "some_body"
-        ret = self.br._serialize_request(body)
-        self.assertIsNotNone(ret)
-        mock_json_dumps.assert_called_once_with(body)
-        mock_json_dumps.return_value.encode.assert_called_once_with("utf-8")
-
-    @patch.object(BarmanRecover, "_deserialize_response", Mock(return_value="test"))
-    @patch("logging.critical")
-    def test__get_request(self, mock_logging):
-        mock_request = self.br.http.request
-
-        # with no error
-        self.assertEqual(self.br._get_request("/some/path"), "test")
-        mock_request.assert_called_once_with("GET", f"{API_URL}/some/path")
-
-        # with MaxRetryError
-        http_error = MaxRetryError(self.br.http, f"{API_URL}/some/path")
-        mock_request.side_effect = http_error
-
-        with self.assertRaises(SystemExit) as exc:
-            self.assertIsNone(self.br._get_request("/some/path"))
-
-        mock_logging.assert_called_once_with("An error occurred while performing an HTTP GET request: %r", http_error)
-        self.assertEqual(exc.exception.code, ExitCode.HTTP_REQUEST_ERROR)
-
-        # with Exception
-        mock_logging.reset_mock()
-        mock_request.side_effect = Exception("Some error.")
-
-        with patch("sys.exit") as mock_sys:
-            with self.assertRaises(Exception):
-                self.assertIsNone(self.br._get_request("/some/path"))
-
-            mock_logging.assert_not_called()
-            mock_sys.assert_not_called()
-
-    @patch.object(BarmanRecover, "_deserialize_response", Mock(return_value="test"))
-    @patch("logging.critical")
-    @patch.object(BarmanRecover, "_serialize_request")
-    def test__post_request(self, mock_serialize, mock_logging):
-        mock_request = self.br.http.request
-
-        # with no error
-        self.assertEqual(self.br._post_request("/some/path", "some body"), "test")
-        mock_serialize.assert_called_once_with("some body")
-        mock_request.assert_called_once_with("POST", f"{API_URL}/some/path", body=mock_serialize.return_value,
-                                             headers={"Content-Type": "application/json"})
-
-        # with HTTPError
-        http_error = MaxRetryError(self.br.http, f"{API_URL}/some/path")
-        mock_request.side_effect = http_error
-
-        with self.assertRaises(SystemExit) as exc:
-            self.assertIsNone(self.br._post_request("/some/path", "some body"))
-
-        mock_logging.assert_called_once_with("An error occurred while performing an HTTP POST request: %r", http_error)
-        self.assertEqual(exc.exception.code, ExitCode.HTTP_REQUEST_ERROR)
-
-        # with Exception
-        mock_logging.reset_mock()
-        mock_request.side_effect = Exception("Some error.")
-
-        with patch("sys.exit") as mock_sys:
-            with self.assertRaises(Exception):
-                self.br._post_request("/some/path", "some body")
-
-            mock_logging.assert_not_called()
-            mock_sys.assert_not_called()
-
-    @patch("logging.critical")
-    @patch.object(BarmanRecover, "_get_request")
-    def test__ensure_api_ok(self, mock_get_request, mock_logging):
-        # API ok
-        mock_get_request.return_value = "OK"
-
-        with patch("sys.exit") as mock_sys:
-            self.assertIsNone(self.br._ensure_api_ok())
-            mock_logging.assert_not_called()
-            mock_sys.assert_not_called()
-
-        # API not ok
-        mock_get_request.return_value = "random"
-
-        with self.assertRaises(SystemExit) as exc:
-            self.assertIsNone(self.br._ensure_api_ok())
-
-        mock_logging.assert_called_once_with("pg-backup-api is not working: %s", "random")
-        self.assertEqual(exc.exception.code, ExitCode.API_NOT_OK)
-
-    @patch("logging.warning")
-    @patch("time.sleep")
-    @patch.object(BarmanRecover, "_post_request")
-    def test__create_recovery_operation(self, mock_post_request, mock_sleep, mock_logging):
-        # well formed response
-        mock_post_request.return_value = {"operation_id": "some_id"}
-        self.assertEqual(self.br._create_recovery_operation(), "some_id")
-        mock_sleep.assert_not_called()
-        mock_logging.assert_not_called()
-        mock_post_request.assert_called_once_with(
-            f"servers/{BARMAN_SERVER}/operations",
-            {
-                "type": "recovery",
-                "backup_id": BACKUP_ID,
-                "remote_ssh_command": SSH_COMMAND,
-                "destination_directory": DATA_DIRECTORY,
-            }
-        )
-
-        # malformed response
-        mock_post_request.return_value = {"operation_idd": "some_id"}
-
-        with self.assertRaises(RetriesExceeded) as exc:
-            self.br._create_recovery_operation()
-
-        self.assertEqual(str(exc.exception),
-                         "Maximum number of retries exceeded for method BarmanRecover._create_recovery_operation.")
-
-        self.assertEqual(mock_sleep.call_count, self.br.max_retries)
-
-        mock_sleep.assert_has_calls([mock.call(self.br.retry_wait)] * self.br.max_retries)
-
-        self.assertEqual(mock_logging.call_count, self.br.max_retries)
-        for i in range(mock_logging.call_count):
-            call_args = mock_logging.call_args_list[i][0]
-            self.assertEqual(len(call_args), 5)
-            self.assertEqual(call_args[0], "Attempt %d of %d on method %s failed with %r.")
-            self.assertEqual(call_args[1], i + 1)
-            self.assertEqual(call_args[2], self.br.max_retries)
-            self.assertEqual(call_args[3], "BarmanRecover._create_recovery_operation")
-            self.assertIsInstance(call_args[4], KeyError)
-            self.assertEqual(call_args[4].args, ('operation_id',))
-
-    @patch("logging.warning")
-    @patch("time.sleep")
-    @patch.object(BarmanRecover, "_get_request")
-    def test__get_recovery_operation_status(self, mock_get_request, mock_sleep, mock_logging):
-        # well formed response
-        mock_get_request.return_value = {"status": "some status"}
-        self.assertEqual(self.br._get_recovery_operation_status("some_id"), "some status")
-        mock_get_request.assert_called_once_with(f"servers/{BARMAN_SERVER}/operations/some_id")
-        mock_sleep.assert_not_called()
-        mock_logging.assert_not_called()
-
-        # malformed response
-        mock_get_request.return_value = {"statuss": "some status"}
-
-        with self.assertRaises(RetriesExceeded) as exc:
-            self.br._get_recovery_operation_status("some_id")
-
-        self.assertEqual(str(exc.exception),
-                         "Maximum number of retries exceeded for method BarmanRecover._get_recovery_operation_status.")
-
-        self.assertEqual(mock_sleep.call_count, self.br.max_retries)
-        mock_sleep.assert_has_calls([mock.call(self.br.retry_wait)] * self.br.max_retries)
-
-        self.assertEqual(mock_logging.call_count, self.br.max_retries)
-        for i in range(mock_logging.call_count):
-            call_args = mock_logging.call_args_list[i][0]
-            self.assertEqual(len(call_args), 5)
-            self.assertEqual(call_args[0], "Attempt %d of %d on method %s failed with %r.")
-            self.assertEqual(call_args[1], i + 1)
-            self.assertEqual(call_args[2], self.br.max_retries)
-            self.assertEqual(call_args[3], "BarmanRecover._get_recovery_operation_status")
-            self.assertIsInstance(call_args[4], KeyError)
-            self.assertEqual(call_args[4].args, ('status',))
-
-    @patch.object(BarmanRecover, "_get_recovery_operation_status")
-    @patch("time.sleep")
-    @patch("logging.info")
-    @patch("logging.critical")
-    @patch.object(BarmanRecover, "_create_recovery_operation")
-    def test_restore_backup(self, mock_create_op, mock_log_critical, mock_log_info, mock_sleep, mock_get_status):
-        # successful fast restore
-        mock_create_op.return_value = "some_id"
-        mock_get_status.return_value = "DONE"
-
-        self.assertTrue(self.br.restore_backup())
-
-        mock_create_op.assert_called_once()
-        mock_get_status.assert_called_once_with("some_id")
-        mock_log_info.assert_called_once_with("Created the recovery operation with ID %s", "some_id")
-        mock_log_critical.assert_not_called()
-        mock_sleep.assert_not_called()
-
-        # successful slow restore
-        mock_create_op.reset_mock()
-        mock_get_status.reset_mock()
-        mock_log_info.reset_mock()
-        mock_get_status.side_effect = ["IN_PROGRESS"] * 20 + ["DONE"]
-
-        self.assertTrue(self.br.restore_backup())
-
-        mock_create_op.assert_called_once()
-
-        self.assertEqual(mock_get_status.call_count, 21)
-        mock_get_status.assert_has_calls([mock.call("some_id")] * 21)
-
-        self.assertEqual(mock_log_info.call_count, 21)
-        mock_log_info.assert_has_calls([mock.call("Created the recovery operation with ID %s", "some_id")]
-                                       + [mock.call("Recovery operation %s is still in progress", "some_id")] * 20)
-
-        mock_log_critical.assert_not_called()
-
-        self.assertEqual(mock_sleep.call_count, 20)
-        mock_sleep.assert_has_calls([mock.call(LOOP_WAIT)] * 20)
-
-        # failed fast restore
-        mock_create_op.reset_mock()
-        mock_get_status.reset_mock()
-        mock_log_info.reset_mock()
-        mock_sleep.reset_mock()
-        mock_get_status.side_effect = None
-        mock_get_status.return_value = "FAILED"
-
-        self.assertFalse(self.br.restore_backup())
-
-        mock_create_op.assert_called_once()
-        mock_get_status.assert_called_once_with("some_id")
-        mock_log_info.assert_called_once_with("Created the recovery operation with ID %s", "some_id")
-        mock_log_critical.assert_not_called()
-        mock_sleep.assert_not_called()
-
-        # failed slow restore
-        mock_create_op.reset_mock()
-        mock_get_status.reset_mock()
-        mock_log_info.reset_mock()
-        mock_sleep.reset_mock()
-        mock_get_status.side_effect = ["IN_PROGRESS"] * 20 + ["FAILED"]
-
-        self.assertFalse(self.br.restore_backup())
-
-        mock_create_op.assert_called_once()
-
-        self.assertEqual(mock_get_status.call_count, 21)
-        mock_get_status.assert_has_calls([mock.call("some_id")] * 21)
-
-        self.assertEqual(mock_log_info.call_count, 21)
-        mock_log_info.assert_has_calls([mock.call("Created the recovery operation with ID %s", "some_id")]
-                                       + [mock.call("Recovery operation %s is still in progress", "some_id")] * 20)
-
-        mock_log_critical.assert_not_called()
-
-        self.assertEqual(mock_sleep.call_count, 20)
-        mock_sleep.assert_has_calls([mock.call(LOOP_WAIT)] * 20)
-
-        # create retries exceeded
-        mock_log_info.reset_mock()
-        mock_sleep.reset_mock()
-        mock_create_op.side_effect = RetriesExceeded
-        mock_get_status.side_effect = None
-
-        with self.assertRaises(SystemExit) as exc:
-            self.assertIsNone(self.br.restore_backup())
-
-        self.assertEqual(exc.exception.code, ExitCode.HTTP_RESPONSE_MALFORMED)
-        mock_log_info.assert_not_called()
-        mock_log_critical.assert_called_once_with("Maximum number of retries exceeded, exiting.")
-        mock_sleep.assert_not_called()
-
-        # get status retries exceeded
-        mock_create_op.reset_mock()
-        mock_create_op.side_effect = None
-        mock_log_critical.reset_mock()
-        mock_log_info.reset_mock()
-        mock_get_status.side_effect = RetriesExceeded
-
-        with self.assertRaises(SystemExit) as exc:
-            self.assertIsNone(self.br.restore_backup())
-
-        self.assertEqual(exc.exception.code, ExitCode.HTTP_RESPONSE_MALFORMED)
-        mock_log_info.assert_called_once_with("Created the recovery operation with ID %s", "some_id")
-        mock_log_critical.assert_called_once_with("Maximum number of retries exceeded, exiting.")
-        mock_sleep.assert_not_called()
-
-
-class TestMain(unittest.TestCase):
-
-    @patch("logging.basicConfig")
-    def test_set_up_logging(self, mock_log_config):
-        log_file = "/path/to/some/file.log"
-        set_up_logging(log_file)
-        mock_log_config.assert_called_once_with(filename=log_file, level=logging.INFO,
-                                                format="%(asctime)s %(levelname)s: %(message)s")
-
-    @patch("logging.critical")
-    @patch("logging.info")
-    @patch("patroni.scripts.barman_recover.set_up_logging")
-    @patch("patroni.scripts.barman_recover.BarmanRecover")
-    @patch("patroni.scripts.barman_recover.ArgumentParser")
-    def test_main(self, mock_arg_parse, mock_br, mock_set_up_log, mock_log_info, mock_log_critical):
-        # successful restore
-        args = MagicMock()
-        mock_arg_parse.return_value.parse_known_args.return_value = (args, None)
-        mock_br.return_value.restore_backup.return_value = True
-
-        with self.assertRaises(SystemExit) as exc:
-            main()
-
-        mock_arg_parse.assert_called_once()
-        mock_set_up_log.assert_called_once_with(args.log_file)
-        mock_br.assert_called_once_with(args.api_url, args.barman_server, args.backup_id, args.ssh_command,
-                                        args.data_directory, args.loop_wait, args.retry_wait, args.max_retries,
-                                        args.cert_file, args.key_file)
-        mock_log_info.assert_called_once_with("Recovery operation finished successfully.")
-        mock_log_critical.assert_not_called()
-        self.assertEqual(exc.exception.code, ExitCode.RECOVERY_DONE)
-
-        # failed restore
-        mock_arg_parse.reset_mock()
-        mock_set_up_log.reset_mock()
-        mock_br.reset_mock()
-        mock_log_info.reset_mock()
-        mock_br.return_value.restore_backup.return_value = False
-
-        with self.assertRaises(SystemExit) as exc:
-            main()
-
-        mock_arg_parse.assert_called_once()
-        mock_set_up_log.assert_called_once_with(args.log_file)
-        mock_br.assert_called_once_with(args.api_url, args.barman_server, args.backup_id, args.ssh_command,
-                                        args.data_directory, args.loop_wait, args.retry_wait, args.max_retries,
-                                        args.cert_file, args.key_file)
-        mock_log_info.assert_not_called()
-        mock_log_critical.assert_called_once_with("Recovery operation failed.")
-        self.assertEqual(exc.exception.code, ExitCode.RECOVERY_FAILED)

tests/test_config.py

@@ -160,12 +160,10 @@ class TestConfig(unittest.TestCase):
     @patch('patroni.config.logger')
     def test__validate_failover_tags(self, mock_logger, mock_get):
         """Ensures that only one of `nofailover` or `failover_priority` can be provided"""
-        config = Config("postgres0.yml")
-
         # Providing one of `nofailover` or `failover_priority` is fine
         for single_param in ({"nofailover": True}, {"failover_priority": 1}, {"failover_priority": 0}):
             mock_get.side_effect = [single_param] * 2
-            self.assertIsNone(config._validate_failover_tags())
+            self.assertIsNone(self.config._validate_failover_tags())
             mock_logger.warning.assert_not_called()
 
         # Providing both `nofailover` and `failover_priority` is fine if consistent
@@ -175,7 +173,7 @@ class TestConfig(unittest.TestCase):
             {"nofailover": "False", "failover_priority": 0}
         ):
             mock_get.side_effect = [consistent_state] * 2
-            self.assertIsNone(config._validate_failover_tags())
+            self.assertIsNone(self.config._validate_failover_tags())
             mock_logger.warning.assert_not_called()
 
         # Providing both inconsistently should log a warning
@@ -186,7 +184,7 @@ class TestConfig(unittest.TestCase):
             {"nofailover": "", "failover_priority": 0}
         ):
             mock_get.side_effect = [inconsistent_state] * 2
-            self.assertIsNone(config._validate_failover_tags())
+            self.assertIsNone(self.config._validate_failover_tags())
             mock_logger.warning.assert_called_once_with(
                 'Conflicting configuration between nofailover: %s and failover_priority: %s.'
                 + ' Defaulting to nofailover: %s',

tests/test_config_generator.py

@@ -142,6 +142,7 @@ class TestGenerateConfig(unittest.TestCase):
                 'noloadbalance': False,
                 'clonefrom': True,
                 'nosync': False,
+                'nostream': False
             }
         }
         patch_config(self.config, conf)

tests/test_consul.py

@@ -4,7 +4,7 @@ import unittest
 from consul import ConsulException, NotFound
 from mock import Mock, PropertyMock, patch
 from patroni.dcs import get_dcs
-from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulAgentService, ConsulInternalError, \
+from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \
     ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError
 from patroni.postgresql.mpp import get_mpp
 from . import SleepException
@@ -160,10 +160,8 @@ class TestConsul(unittest.TestCase):
         self.c.set_ttl(20)
         self.c._do_refresh_session = Mock()
         self.assertFalse(self.c.take_leader())
-        with patch('time.time', Mock(side_effect=[0, 100])):
-            self.assertRaises(ConsulError, self.c.take_leader)
-        with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 0, 100])):
-            self.assertRaises(ConsulError, self.c.take_leader)
+        with patch('time.time', Mock(side_effect=[0, 0, 0, 100, 100, 100])):
+            self.assertFalse(self.c.take_leader())
 
     @patch.object(consul.Consul.KV, 'put', Mock(return_value=True))
     def test_set_failover_value(self):
@@ -245,8 +243,8 @@ class TestConsul(unittest.TestCase):
     def test_set_history_value(self):
         self.assertTrue(self.c.set_history_value('{}'))
 
-    @patch.object(ConsulAgentService, 'register', Mock(side_effect=(False, True, True, True)))
-    @patch.object(ConsulAgentService, 'deregister', Mock(return_value=True))
+    @patch.object(consul.Consul.Agent.Service, 'register', Mock(side_effect=(False, True, True, True)))
+    @patch.object(consul.Consul.Agent.Service, 'deregister', Mock(return_value=True))
     def test_update_service(self):
         d = {'role': 'replica', 'api_url': 'http://a/t', 'conn_url': 'pg://c:1', 'state': 'running'}
         self.assertIsNone(self.c.update_service({}, {}))
@@ -277,7 +275,7 @@ class TestConsul(unittest.TestCase):
 
         # Changing register_service from True to False calls deregister()
         self.c.reload_config({'consul': {'register_service': False}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10})
-        with patch.object(ConsulAgentService, 'deregister') as mock_deregister:
+        with patch('consul.Consul.Agent.Service.deregister') as mock_deregister:
             self.c.touch_member(d)
             mock_deregister.assert_called_once()
 
@@ -285,31 +283,31 @@ class TestConsul(unittest.TestCase):
 
         # register_service staying False between reloads does not call deregister()
         self.c.reload_config({'consul': {'register_service': False}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10})
-        with patch.object(ConsulAgentService, 'deregister') as mock_deregister:
+        with patch('consul.Consul.Agent.Service.deregister') as mock_deregister:
             self.c.touch_member(d)
             self.assertFalse(mock_deregister.called)
 
         # Changing register_service from False to True calls register()
         self.c.reload_config({'consul': {'register_service': True}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10})
-        with patch.object(HTTPClient, 'put', create=True) as mock_put:
+        with patch('consul.Consul.Agent.Service.register') as mock_register:
             self.c.touch_member(d)
-            mock_put.assert_called_once()
+            mock_register.assert_called_once()
 
         # register_service staying True between reloads does not call register()
         self.c.reload_config({'consul': {'register_service': True}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10})
-        with patch.object(ConsulAgentService, 'register') as mock_register:
+        with patch('consul.Consul.Agent.Service.register') as mock_register:
             self.c.touch_member(d)
             self.assertFalse(mock_deregister.called)
 
         # register_service staying True between reloads does calls register() if other service data has changed
         self.c.reload_config({'consul': {'register_service': True}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10})
-        with patch.object(ConsulAgentService, 'register') as mock_register:
+        with patch('consul.Consul.Agent.Service.register') as mock_register:
             self.c.touch_member(d)
             mock_register.assert_called_once()
 
         # register_service staying True between reloads does calls register() if service_tags have changed
         self.c.reload_config({'consul': {'register_service': True, 'service_tags': ['foo']}, 'loop_wait': 10,
                               'ttl': 30, 'retry_timeout': 10})
-        with patch.object(ConsulAgentService, 'register') as mock_register:
+        with patch('consul.Consul.Agent.Service.register') as mock_register:
             self.c.touch_member(d)
             mock_register.assert_called_once()

tests/test_etcd3.py

@@ -7,7 +7,7 @@ from mock import Mock, PropertyMock, patch
 from patroni.dcs import get_dcs
 from patroni.dcs.etcd import DnsCachingResolver
 from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \
-    Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \
+    Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \
     Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode
 from patroni.postgresql.mpp import get_mpp
 from threading import Thread
@@ -166,12 +166,14 @@ class TestPatroniEtcd3Client(BaseTestEtcd3):
             retry = self.etcd3._retry.copy()
             with patch('time.time', Mock(side_effect=[0, 10, 20, 30, 40])):
                 self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry)
+            with patch('time.time', Mock(side_effect=[0, 10])):
+                self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
             self.client.username = None
-            self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
+            self.client._reauthenticate = False
             retry = self.etcd3._retry.copy()
             self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry)
             mock_urlopen.return_value.content = '{"code":3,"error":"etcdserver: revision of auth store is old"}'
-            self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
+            self.client._reauthenticate = False
             self.assertRaises(AuthOldRevision, retry, self.client.deleteprefix, 'foo', retry=retry)
 
     def test__handle_server_response(self):
@@ -271,8 +273,8 @@ class TestEtcd3(BaseTestEtcd3):
 
     def test_attempt_to_acquire_leader(self):
         self.assertFalse(self.etcd3.attempt_to_acquire_leader())
-        with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 100, 200])):
-            self.assertRaises(Etcd3Error, self.etcd3.attempt_to_acquire_leader)
+        with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 100, 200, 300])):
+            self.assertFalse(self.etcd3.attempt_to_acquire_leader())
         with patch('time.time', Mock(side_effect=[0, 100, 200, 300, 400])):
             self.assertRaises(Etcd3Error, self.etcd3.attempt_to_acquire_leader)
         with patch.object(PatroniEtcd3Client, 'put', Mock(return_value=False)):

tests/test_ha.py

@@ -151,6 +151,7 @@ zookeeper:
         self.api.connection_string = 'http://127.0.0.1:8008'
         self.clonefrom = None
         self.nosync = False
+        self.nostream = False
         self.scheduled_restart = {'schedule': future_restart_time,
                                   'postmaster_start_time': str(postmaster_start_time)}
         self.watchdog = Watchdog(self.config)
@@ -255,8 +256,6 @@ class TestHa(PostgresInit):
         self.p.data_directory_empty = true
         self.ha.cluster = get_cluster_not_initialized_without_leader(
             cluster_config=ClusterConfig(1, {"standby_cluster": {"port": 5432}}, 1))
-        global_config.update(self.ha.cluster)
-        self.ha.cluster = get_cluster_not_initialized_without_leader(cluster_config=ClusterConfig(0, {}, 0))
         self.assertEqual(self.ha.run_cycle(), 'trying to bootstrap a new standby leader')
 
     def test_bootstrap_waiting_for_standby_leader(self):
@@ -322,7 +321,6 @@ class TestHa(PostgresInit):
             self.ha.state_handler.cancellable._process = Mock()
             self.ha._crash_recovery_started -= 600
             self.ha.cluster.config.data.update({'maximum_lag_on_failover': 10})
-            global_config.update(self.ha.cluster)
             self.assertEqual(self.ha.run_cycle(), 'terminated crash recovery because of startup timeout')
 
     @patch.object(Rewind, 'ensure_clean_shutdown', Mock())
@@ -474,6 +472,11 @@ class TestHa(PostgresInit):
         self.p.is_primary = false
         self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS')
 
+    def test_get_node_to_follow_nostream(self):
+        self.ha.patroni.nostream = True
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.assertEqual(self.ha._get_node_to_follow(self.ha.cluster), None)
+
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_follow(self):
         self.p.is_primary = false
@@ -770,7 +773,6 @@ class TestHa(PostgresInit):
         with patch('patroni.ha.logger.info') as mock_info:
             self.ha.fetch_node_status = get_node_status(wal_position=1)
             self.ha.cluster.config.data.update({'maximum_lag_on_failover': 5})
-            global_config.update(self.ha.cluster)
             self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
             self.assertEqual(mock_info.call_args_list[0][0], ('Member %s exceeds maximum replication lag', 'leader'))
 
@@ -1276,7 +1278,6 @@ class TestHa(PostgresInit):
         self.p.is_running = false
         self.ha.cluster = get_cluster_initialized_with_leader(sync=(self.p.name, 'other'))
         self.ha.cluster.config.data.update({'synchronous_mode': True, 'primary_start_timeout': 0})
-        global_config.update(self.ha.cluster)
         self.ha.has_lock = true
         self.ha.update_lock = true
         self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
@@ -1385,7 +1386,6 @@ class TestHa(PostgresInit):
         mock_set_sync.reset_mock()
         self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet()))
         self.ha.cluster.config.data['synchronous_mode_strict'] = True
-        global_config.update(self.ha.cluster)
         self.ha.run_cycle()
         mock_set_sync.assert_called_once_with(CaseInsensitiveSet('*'))
 

tests/test_kubernetes.py

@@ -80,7 +80,7 @@ def mock_namespaced_kind(*args, **kwargs):
 
 
 def mock_load_k8s_config(self, *args, **kwargs):
-    self._server = ''
+    self._server = 'http://localhost'
 
 
 class TestK8sConfig(unittest.TestCase):
@@ -242,6 +242,7 @@ class BaseTestKubernetes(unittest.TestCase):
         self.k.get_cluster()
 
 
+@patch('urllib3.PoolManager.request', Mock())
 @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_namespaced_kind, create=True)
 class TestKubernetesConfigMaps(BaseTestKubernetes):
 
@@ -374,6 +375,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
         mock_warning.assert_called_once()
 
 
+@patch('urllib3.PoolManager.request', Mock())
 class TestKubernetesEndpointsNoPodIP(BaseTestKubernetes):
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True)
     def setUp(self, config=None):
@@ -388,6 +390,7 @@ class TestKubernetesEndpointsNoPodIP(BaseTestKubernetes):
         self.assertEqual(args[2].subsets[0].addresses[0].ip, '10.0.0.1')
 
 
+@patch('urllib3.PoolManager.request', Mock())
 class TestKubernetesEndpoints(BaseTestKubernetes):
 
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True)
@@ -478,6 +481,7 @@ def mock_watch(*args):
     return urllib3.HTTPResponse()
 
 
+@patch('urllib3.PoolManager.request', Mock())
 class TestCacheBuilder(BaseTestKubernetes):
 
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)

tests/test_patroni.py

@@ -249,6 +249,16 @@ class TestPatroni(unittest.TestCase):
         self.p.tags['nosync'] = None
         self.assertFalse(self.p.nosync)
 
+    def test_nostream(self):
+        self.p.tags['nostream'] = 'True'
+        self.assertTrue(self.p.nostream)
+        self.p.tags['nostream'] = 'None'
+        self.assertFalse(self.p.nostream)
+        self.p.tags['nostream'] = 'foo'
+        self.assertFalse(self.p.nostream)
+        self.p.tags['nostream'] = ''
+        self.assertFalse(self.p.nostream)
+
     @patch.object(Thread, 'join', Mock())
     def test_shutdown(self):
         self.p.api.shutdown = Mock(side_effect=Exception)

tests/test_postgresql.py

@@ -7,6 +7,7 @@ import time
 
 from copy import deepcopy
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
+from pathlib import Path
 
 import patroni.psycopg as psycopg
 
@@ -363,7 +364,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'start', Mock())
     def test_follow(self):
         self.p.call_nowait(CallbackAction.ON_START)
-        m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}})
+        m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'foo,bar'}})
         self.p.follow(m)
         with patch.object(Postgresql, 'ensure_major_version_is_known', Mock(return_value=False)):
             self.assertIsNone(self.p.follow(m))
@@ -1064,7 +1065,7 @@ class TestPostgresql(BaseTestPostgresql):
     def test__read_postgres_gucs_validators_file(self):
         # raise exception
         with self.assertRaises(InvalidGucValidatorsFile) as exc:
-            _read_postgres_gucs_validators_file('random_file.yaml')
+            _read_postgres_gucs_validators_file(Path('random_file.yaml'))
         self.assertEqual(
             str(exc.exception),
             "Unexpected issue while reading parameters file `random_file.yaml`: `[Errno 2] No such file or directory: "
@@ -1073,17 +1074,32 @@ class TestPostgresql(BaseTestPostgresql):
 
     def test__load_postgres_gucs_validators(self):
         # log messages
-        with patch('os.walk', Mock(return_value=iter([('.', [], ['file.txt', 'random.yaml'])]))), \
-             patch('patroni.postgresql.validator.logger.info') as mock_info, \
+        file1_attrs = {'is_file.return_value': True, 'is_dir.return_value': False}
+        file1_mock = MagicMock(**file1_attrs)
+        file1_mock.name = '__init__.py'
+        file2_attrs = {'is_file.return_value': False, 'is_dir.return_value': True, 'iterdir.return_value': []}
+        file2_mock = MagicMock(**file2_attrs)
+        file2_mock.name = '__pycache__'
+        file3_attrs = {'is_file.return_value': True, 'is_dir.return_value': False}
+        file3_mock = MagicMock(**file3_attrs)
+        file3_mock.name = file3_mock.__str__.return_value = 'random.yaml'
+        file3_mock.open.side_effect = FileNotFoundError('[Errno 2] No such file or directory: random.yaml')
+        file4_attrs = {'is_file.return_value': True, 'is_dir.return_value': False}
+        file4_mock = MagicMock(**file4_attrs)
+        file4_mock.name = 'file.txt'
+        dir_attrs = {'name': 'available_parameters', 'is_file.return_value': False, 'is_dir.return_value': True}
+        dir_mock = MagicMock(**dir_attrs)
+        dir_mock.iterdir.return_value = [file1_mock, file2_mock, file3_mock, file4_mock]
+        with patch('patroni.postgresql.available_parameters.conf_dir', dir_mock), \
+             patch('patroni.postgresql.available_parameters.logger.info') as mock_info, \
              patch('patroni.postgresql.validator.logger.warning') as mock_warning:
             _load_postgres_gucs_validators()
-            mock_info.assert_called_once_with('Ignored a non-YAML file found under `available_parameters` directory: '
-                                              '`%s`.', os.path.join('.', 'file.txt'))
+            mock_info.assert_called_once_with('Ignored a non-YAML file found under `%s` '
+                                              'directory: `%s`.', 'available_parameters', file4_mock)
             mock_warning.assert_called_once()
             self.assertIn(
-                "Unexpected issue while reading parameters file `{0}`: `[Errno 2] No such file or "
-                "directory:".format(os.path.join('.', 'random.yaml')),
-                mock_warning.call_args[0][0]
+                "Unexpected issue while reading parameters file `random.yaml`: `[Errno 2] No such file or "
+                "directory:", mock_warning.call_args[0][0]
             )
 
 

tests/test_rewind.py

@@ -98,7 +98,8 @@ class TestRewind(BaseTestPostgresql):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
     @patch.object(CancellableSubprocess, 'call', mock_cancellable_call)
-    @patch.object(Postgresql, 'checkpoint', side_effect=['', '1'],)
+    @patch.object(Postgresql, 'get_guc_value', Mock(return_value=''))
+    @patch.object(Postgresql, 'checkpoint', side_effect=['', '1'])
     @patch.object(Postgresql, 'stop', Mock(return_value=False))
     @patch.object(Postgresql, 'start', Mock())
     def test_execute(self, mock_checkpoint):

tests/test_slots.py

@@ -124,6 +124,68 @@ class TestSlotsHandler(BaseTestPostgresql):
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])]
             self.assertEqual(self.p.slots(), {})
 
+    def test_nostream_slot_processing(self):
+        config = ClusterConfig(
+            1, {'slots': {'foo': {'type': 'logical', 'database': 'a', 'plugin': 'b'}, 'bar': {'type': 'physical'}}}, 1)
+        nostream_node = Member(0, 'test-2', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+            'tags': {'nostream': 'True'}
+        })
+        cascade_node = Member(0, 'test-3', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+            'tags': {'replicatefrom': 'test-2'}
+        })
+        stream_node = Member(0, 'test-4', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres'})
+        cluster = Cluster(
+            True, config, self.leader, Status.empty(),
+            [self.leadermem, nostream_node, cascade_node, stream_node], None, SyncState.empty(), None, None)
+        global_config.update(cluster)
+
+        # sanity for primary
+        self.p.name = self.leadermem.name
+        self.assertEqual(
+            cluster._get_permanent_slots(self.p, self.leadermem, 'primary'),
+            {'foo': {'type': 'logical', 'database': 'a', 'plugin': 'b'}, 'bar': {'type': 'physical'}})
+        self.assertEqual(
+            cluster._get_members_slots(self.p.name, 'primary'),
+            {'test_4': {'type': 'physical'}})
+
+        # nostream node must not have slot on primary
+        self.p.name = nostream_node.name
+        # permanent logical slots are not allowed on nostream node
+        self.assertEqual(
+            cluster._get_permanent_slots(self.p, nostream_node, 'replica'),
+            {'bar': {'type': 'physical'}})
+        self.assertEqual(
+            cluster.get_slot_name_on_primary(self.p.name, nostream_node),
+            None)
+
+        # check cascade member-slot existence on nostream node
+        self.assertEqual(
+            cluster._get_members_slots(nostream_node.name, 'replica'),
+            {'test_3': {'type': 'physical'}})
+
+        # cascade also does not entitled to have logical slot on itself ...
+        self.p.name = cascade_node.name
+        self.assertEqual(
+            cluster._get_permanent_slots(self.p, cascade_node, 'replica'),
+            {'bar': {'type': 'physical'}})
+        # ... and member-slot on primary
+        self.assertEqual(
+            cluster.get_slot_name_on_primary(self.p.name, cascade_node),
+            None)
+
+        # simple replica must have every permanent slot ...
+        self.p.name = stream_node.name
+        self.assertEqual(
+            cluster._get_permanent_slots(self.p, stream_node, 'replica'),
+            {'foo': {'type': 'logical', 'database': 'a', 'plugin': 'b'}, 'bar': {'type': 'physical'}})
+        # ... and member-slot on primary
+        self.assertEqual(
+            cluster.get_slot_name_on_primary(self.p.name, stream_node),
+            'test_4')
+
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')

tests/test_validator.py

@@ -103,7 +103,8 @@ config = {
         "nofailover": False,
         "clonefrom": False,
         "noloadbalance": False,
-        "nosync": False
+        "nosync": False,
+        "nostream": False
     }
 }
 

typings/consul/base.pyi

@@ -1,14 +1,10 @@
-from typing import Any, Callable, Dict, List, NamedTuple, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 class ConsulException(Exception): ...
 class NotFound(ConsulException): ...
-class CB:
-    @classmethod
-    def bool(klass) -> Callable[[NamedTuple], bool]: ...
 class Check:
     @classmethod
     def http(klass, url: str, interval: str, timeout: Optional[str] = None, deregister: Optional[str] = None) -> Dict[str, str]: ...
 class Consul:
-    token: Optional[str]
     http: Any
     agent: 'Consul.Agent'
     session: 'Consul.Session'
@@ -21,9 +17,7 @@ class Consul:
         service: 'Consul.Agent.Service'
         def self(self) -> Dict[str, Dict[str, Any]]: ...
         class Service:
-            agent: 'Consul'
-            def __init__(self, agent: 'Consul') -> None: ..
-            def register(self, name: str, service_id: Optional[str] = None, address: Optional[str] = None, port: Optional[int] = None, tags: Optional[List[str]] = None, check: Optional[Dict[str, str]] = None, token: Optional[str] = None, enable_tag_override: bool = False) -> bool: ...
+            def register(self, name: str, service_id=..., address=..., port=..., tags=..., check=..., token=..., script=..., interval=..., ttl=..., http=..., timeout=..., enable_tag_override=...) -> bool: ...
             def deregister(self, service_id: str) -> bool: ...
     class Session:
         def create(self, name: Optional[str] = None, node: Optional[str] = [], checks: Optional[List[str]]=None, lock_delay: float = 15, behavior: str = 'release', ttl: Optional[int] = None, dc: Optional[str] = None) -> str: ...