Repository: zalando/patroni
Pull request #2931 (2023-10-26 16:15:58)

Add a contrib script for remote Barman recovery
This PR adds a contrib script, which can be used as a custom bootstrap method, or as a custom create replica method.

The script communicates with the pg-backup-api on the Barman node so Patroni is able to restore a Barman backup remotely.

The `--help` option of the script, along with the script docstring, should provide some context on how to use fill its parameters.

Patroni docs were updated accordingly to share examples about how to configure the script as a custom bootstrap method, or as a custom create replica method.

References: PAT-216.

coveralls: ## Pull Request Test Coverage Report for [Build 6656892816](https://coveralls.io/builds/63565769)

* **45** of **143**   **(31.47%)**  changed or added relevant lines in **1** file are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage decreased (**-0.7%**) to **99.132%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [patroni/scripts/barman_recover.py](https://coveralls.io/builds/63565769/source?filename=patroni%2Fscripts%2Fbarman_recover.py#L79) | 45 | 143 | 31.47%
<!-- | **Total:** | **45** | **143** | **31.47%** | -->


|  Totals | [![Coverage Status](https://coveralls.io/builds/63565769/badge)](https://coveralls.io/builds/63565769) |
| :-- | --: |
| Change from base [Build 6641691394](https://coveralls.io/builds/63532880): |  -0.7% |
| Covered Lines: | 13249 |
| Relevant Lines: | 13365 |

---
##### 💛  - [Coveralls](https://coveralls.io)

barthisrael: Also missing unit tests 🤦 
barthisrael: I started writing them today, will finish the unit tests by tomorrow.
barthisrael: The PR is now ready for review, with proper changes to `setup.py` and with unit test :)
barthisrael: Codacy is still not happy about my check in regards to URI scheme :(
Will I need to add an exception raising right before `urlopen` calls?
CyberDem0n: I think Codacy will shut up when you switch to urllib3. At least it doesn't complain about other parts.
barthisrael: All points might be addressed at this point. Let's see if Codacy calms down :)

Patch:
diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst
index bd928af9..ad7c46b8 100644
--- a/docs/ENVIRONMENT.rst
+++ b/docs/ENVIRONMENT.rst
@@ -85,6 +85,7 @@ ZooKeeper
 -  **PATRONI\_ZOOKEEPER\_KEY\_PASSWORD**: (optional) The client key password.
 -  **PATRONI\_ZOOKEEPER\_VERIFY**: (optional) Whether to verify certificate or not. Defaults to ``true``.
 -  **PATRONI\_ZOOKEEPER\_SET\_ACLS**: (optional) If set, configure Kazoo to apply a default ACL to each ZNode that it creates. ACLs will assume 'x509' schema and should be specified as a dictionary with the principal as the key and one or more permissions as a list in the value.  Permissions may be one of ``CREATE``, ``READ``, ``WRITE``, ``DELETE`` or ``ADMIN``.  For example, ``set_acls: {CN=principal1: [CREATE, READ], CN=principal2: [ALL]}``.
+-  **PATRONI\_ZOOKEEPER\_AUTH\_DATA**: (optional) Authentication credentials to use for the connection. Should be a dictionary in the form that `scheme` is the key and `credential` is the value. Defaults to empty dictionary.
 
 .. note::
     It is required to install ``kazoo>=2.6.0`` to support SSL.
diff --git a/docs/contributing_guidelines.rst b/docs/contributing_guidelines.rst
index 805f2fa8..48d2b77a 100644
--- a/docs/contributing_guidelines.rst
+++ b/docs/contributing_guidelines.rst
@@ -3,11 +3,15 @@
 Contributing guidelines
 =======================
 
+.. _chatting:
+
 Chatting
 --------
 
 If you have a question, looking for an interactive troubleshooting help or want to chat with other Patroni users, join us on channel `#patroni <https://postgresteam.slack.com/archives/C9XPYG92A>`__ in the `PostgreSQL Slack <https://pgtreats.info/slack-invite>`__.
 
+.. _reporting_bugs:
+
 Reporting bugs
 --------------
 
diff --git a/docs/faq.rst b/docs/faq.rst
new file mode 100644
index 00000000..c56bf0ce
--- /dev/null
+++ b/docs/faq.rst
@@ -0,0 +1,329 @@
+.. _faq:
+
+FAQ
+===
+
+In this section you will find answers for the most frequently asked questions about Patroni.
+Each sub-section attempts to focus on different kinds of questions.
+
+We hope that this helps you to clarify most of your questions.
+If you still have further concerns or find yourself facing an unexpected issue, please refer to :ref:`chatting` and :ref:`reporting_bugs` for instructions on how to get help or report issues.
+
+Comparison with other HA solutions
+----------------------------------
+
+Why does Patroni require a separate cluster of DCS nodes while other solutions like ``repmgr`` do not?
+    There are different ways of implementing HA solutions, each of them with their pros and cons.
+
+    Software like ``repmgr`` performs communication among the nodes to decide when actions should be taken.
+
+    Patroni on the other hand relies on the state stored in the DCS. The DCS acts as a source of truth for Patroni to decide what it should do.
+
+    While having a separate DCS cluster can make you bloat your architecture, this approach also makes it less likely for split-brain scenarios to happen in your Postgres cluster.
+
+What is the difference between Patroni and other HA solutions in regards to Postgres management?
+    Patroni does not just manage the high availability of the Postgres cluster but also manages Postgres itself.
+
+    If Postgres nodes do not exist yet, it takes care of bootstrapping the primary and the standby nodes, and also manages Postgres configuration of the nodes. If the Postgres nodes already exist, Patroni will take over management of the cluster.
+
+    Besides the above, Patroni also has self-healing capabilities. In other words, if a primary node fails, Patroni will not only fail over to a replica, but also attempt to rejoin the former primary as a replica of the new primary. Similarly, if a replica fails, Patroni will attempt to rejoin that replica.
+
+    That is way we call Patroni as a "template for HA solutions". It goes further than just managing physical replication: it manages Postgres as a whole.
+
+DCS
+---
+
+Can I use the same ``etcd`` cluster to store data from two or more Patroni clusters?
+    Yes, you can!
+
+    Information about a Patroni cluster is stored in the DCS under a path prefixed with the ``namespace`` and ``scope`` Patroni settings.
+
+    As long as you do not have conflicting namespace and scope across different Patroni clusters, you should be able to use the same DCS cluster to store information from multiple Patroni clusters.
+
+What occurs if I attempt to use the same combination of ``namespace`` and ``scope`` for different Patroni clusters that point to the same DCS cluster?
+    The second Patroni cluster that attempts to use the same ``namespace`` and ``scope`` will not be able to manage Postgres because it will find information related with that same combination in the DCS, but with an incompatible Postgres system identifier.
+    The mismatch on the system identifier causes Patroni to abort the management of the second cluster, as it assumes that refers to a different cluster and that the user has misconfigured Patroni.
+
+    Make sure to use different ``namespace`` / ``scope`` when dealing with different Patroni clusters that share the same DCS cluster.
+
+What occurs if I lose my DCS cluster?
+    The DCS is used to store basically status and the dynamic configuration of the Patroni cluster.
+
+    They very first consequence is that all the Patroni clusters that rely on that DCS will go to read-only mode -- unless :ref:`dcs_failsafe_mode` is enabled.
+
+What should I do if I lose my DCS cluster?
+    There are three possible outcomes upon losing your DCS cluster:
+
+    1. The DCS cluster is fully recovered: this requires no action from the Patroni side. Once the DCS cluster is recovered, Patroni should be able to recover too;
+    2. The DCS cluster is re-created in place, and the endpoints remain the same. No changes are required on the Patroni side;
+    3. A new DCS cluster is created with different endpoints. You will need to update the DCS endpoints in the Patroni configuration of each Patroni node.
+
+    If you face scenario ``2.`` or ``3.`` Patroni will take care of creating the status information again based on the current status of the cluster, and recreate the dynamic configuration on the DCS based on a backup file named ``patroni.dynamic.json`` which is stored inside the Postgres data directory of each member of the Patroni cluster.
+
+What occurs if I lose majority in my DCS cluster?
+    The DCS will become unresponsive, which will cause Patroni to demote the current read/write Postgres node.
+
+    Remember: Patroni relies on the state of the DCS to take actions on the cluster.
+
+    You can use the :ref:`dcs_failsafe_mode` to alleviate that situation.
+
+patronictl
+----------
+
+Do I need to run :ref:`patronictl` in the Patroni host?
+    No, you do not need to do that.
+
+    Running :ref:`patronictl` in the Patroni host is handy if you have access to the Patroni host because you can use the very same configuration file from the ``patroni`` agent for the :ref:`patronictl` application.
+
+    However, :ref:`patronictl` is basically a client and it can be executed from remote machines. You just need to provide it with enough configuration so it can reach the DCS and the REST API of the Patroni member(s).
+
+Why did the information from one of my Patroni members disappear from the output of :ref:`patronictl_list` command?
+    Information shown by :ref:`patronictl_list` is based on the contents of the DCS.
+
+    If information about a member disappeared from the DCS it is very likely that the Patroni agent on that node is not running anymore, or it is not able to communicate with the DCS.
+
+    As the member is not able to update the information, the information eventually expires from the DCS, and consequently the member is not shown anymore in the output of :ref:`patronictl_list`.
+
+Why is the information about one of my Patroni members not up-to-date in the output of :ref:`patronictl_list` command?
+    Information shown by :ref:`patronictl_list` is based on the contents of the DCS.
+
+    By default, that information is updated by Patroni roughly every ``loop_wait`` seconds.
+    In other words, even if everything is normally functional you may still see a "delay" of up to ``loop_wait`` seconds in the information stored in the DCS.
+
+    Be aware that that is not a rule, though. Some operations performed by Patroni cause it to immediately update the DCS information.
+
+Configuration
+-------------
+
+What is the difference between dynamic configuration and local configuration?
+    Dynamic configuration (or global configuration) is the configuration stored in the DCS, and which is applied to all members of the Patroni cluster.
+    This is primarily where you should store your configuration.
+
+    Settings that are specific to a node, or settings that you would like to overwrite the global configuration with, you should set only on the desired Patroni member as a local configuration.
+    That local configuration can be specified either through the configuration file or through environment variables.
+
+    See more in :ref:`patroni_configuration`.
+
+What are the types of configuration in Patroni, and what is the precedence?
+    The types are:
+
+    * Dynamic configuration: applied to all members;
+    * Local configuration: applied to the local member, overrides dynamic configuration;
+    * Environment configuration: applied to the local member, overrides both dynamic and local configuration.
+
+    **Note:** some Postgres GUCs can only be set globally, i.e., through dynamic configuration. Besides that, there are GUCs which Patroni enforces a hard-coded value.
+
+    See more in :ref:`patroni_configuration`.
+
+Is there any facility to help me create my Patroni configuration file?
+    Yes, there is.
+
+    You can use ``patroni --generate-sample-config`` or ``patroni --generate-config`` commands to generate a sample Patroni configuration or a Patroni configuration based on an existing Postgres instance, respectively.
+
+    Please refer to :ref:`generate_sample_config` and :ref:`generate_config` for more details.
+
+I changed my parameters under ``bootstrap.dcs`` configuration but Patroni is not applying the changes to the cluster members. What is wrong?
+    The values configured under ``bootstrap.dcs`` are only used when bootstrapping a fresh cluster. Those values will be written to the DCS during the bootstrap.
+
+    After the bootstrap phase finishes, you will only be able to change the dynamic configuration through the DCS.
+
+    Refer to the next question for more details.
+
+How can I change my dynamic configuration?
+    You need to change the configuration in the DCS. That is accomplished either through:
+
+    * :ref:`patronictl_edit_config`; or
+    * A ``PATCH`` request to :ref:`config_endpoint`.
+
+How can I change my local configuration?
+    You need to change the configuration file of the corresponding Patroni member and signal the Patroni agent with ``SIHGUP``. You can do that using either of these approaches:
+
+    * Send a ``POST`` request to the REST API :ref:`reload_endpoint`; or
+    * Run :ref:`patronictl_reload`; or
+    * Locally signal the Patroni process with ``SIGHUP``:
+
+        * If you started Patroni through systemd, you can use the command ``systemctl reload PATRONI_UNIT.service``, ``PATRONI_UNIT`` being the name of the Patroni service; or
+        * If you started Patroni through other means, you will need to identify the ``patroni`` process and run ``kill -s HUP PID``, ``PID`` being the process ID of the ``patroni`` process.
+
+    **Note:** there are cases where a reload through the :ref:`patronictl_reload` may not work:
+
+    * Expired REST API certificates: you can mitigate that by using the ``-k`` option of the :ref:`patronictl`;
+    * Wrong credentials: for example when changing ``restapi`` or ``ctl`` credentials in the configuration file, and using that same configuration file for Patroni and :ref:`patronictl`.
+
+How can I change my environment configuration?
+    The environment configuration is only read by Patroni during startup.
+
+    With that in mind, if you change the environment configuration you will need to restart the corresponding Patroni agent.
+
+    Take care to not cause a failover in the cluster! You might be interested in checking :ref:`patronictl_pause`.
+
+What occurs if I change a Postgres GUC that requires a reload?
+    When you change the dynamic or the local configuration as explained in the previous questions, Patroni will take care of reloading the Postgres configuration for you.
+
+What occurs if I change a Postgres GUC that requires a restart?
+    Patroni will mark the affected members with a flag of ``pending restart``.
+
+    It is up to you to determine when and how to restart the members. That can be accomplished either through:
+
+    * :ref:`patronictl_restart`; or
+    * A ``POST`` request to :ref:`restart_endpoint`.
+
+    **Note:** some Postgres GUCs require a special management in terms of the order for restarting the Postgres nodes. Refer to :ref:`shared_memory_gucs` for more details.
+
+What is the difference between ``etcd`` and ``etcd3`` in Patroni configuration?
+    ``etcd`` uses the API version 2 of ``etcd``, while ``etcd3`` uses the API version 3 of ``etcd``.
+
+    Be aware that information stored by the API version 2 is not manageable by API version 3 and vice-versa.
+
+    We recommend that you configure ``etcd3`` instead of ``etcd`` because:
+
+    * API version 2 is disabled by default from Etcd v3.4 onward;
+    * API version 2 will be completely removed on Etcd v3.6.
+
+I have ``use_slots`` enabled in my Patroni configuration, but when a cluster member goes offline for some time, the replication slot used by that member is dropped on the upstream node. What can I do to avoid that issue?
+    You can configure a permanent physical replication slot for the members.
+
+    Since Patroni ``3.2.0`` it is now possible to have member slots as permanent slots managed by Patroni.
+
+    Patroni will create the permanent physical slots on all nodes, and make sure to not remove the slots, as well as to advance the slots' LSN on all nodes according to the LSN that has been consumed by the member.
+
+    Later, if you decide to remove the corresponding member, it's **your responsability** to adjust the permanent slots configuration, otherwise Patroni will keep the slots around forever.
+
+    **Note:** on Patroni older than ``3.2.0`` you could still have member slots configured as permanent physical slots, however they would be managed only on the current leader. That is, in case of failover/switchover these slots would be created on the new leader, but that wouldn't guarantee that it had all WAL segments for the absent node.
+
+    **Note:** even with Patroni ``3.2.0`` there might be a small race condition. In the very beginning, when the slot is created on the replica it could be ahead of the same slot on the leader and in case if nobody is consuming the slot there is still a chance that some files could be missing after failover. With that in mind, it is recommended that you configure continuous archiving, which makes it possible to restore required WALs or perform PITR.
+
+What is the difference between ``loop_wait``, ``retry_timeout`` and ``ttl``?
+    Patroni performs what we call a HA cycle from time to time. On each HA cycle it takes care of performing a series of checks on the cluster to determine its healthiness, and depending on the status it may take actions, like failing over to a standby.
+
+    ``loop_wait`` determines for how long, in seconds, Patroni should sleep before performing a new cycle of HA checks.
+
+    ``retry_timeout`` sets the timeout for retry operations on the DCS and on Postgres. For example: if the DCS is unresponsive for more than ``retry_timeout`` seconds, Patroni might demote the primary node as a security action.
+
+    ``ttl`` sets the lease time on the ``leader`` lock in the DCS. If the current leader of the cluster is not able to renew the lease during its HA cycles for longer than ``ttl``, then the lease will expire and that will trigger a ``leader race`` in the cluster.
+
+    **Note:** when modifying these settings, please keep in mind that Patroni enforces the rule and minimal values described in :ref:`dynamic_configuration` section of the docs.
+
+Postgres management
+-------------------
+
+Can I change Postgres GUCs directly in Postgres configuration?
+    You can, but you should avoid that.
+
+    Postgres configuration is managed by Patroni, and attempts to edit the configuration files may end up being frustrated by Patroni as it may eventually overwrite them.
+
+    There are a few options available to overcome the management performed by Patroni:
+
+    * Change Postgres GUCs through ``$PGDATA/postgresql.base.conf``; or
+    * Define a ``postgresql.custom_conf`` which will be used instead of ``postgresql.base.conf`` so you can manage that externally; or
+    * Change GUCs using ``ALTER SYSTEM`` / ``ALTER DATABASE`` / ``ALTER USER``.
+
+    You can find more information about that in the section :ref:`important_configuration_rules`.
+
+    In any case we recommend that you manage all the Postgres configuration through Patroni. That will centralize the management and make it easier to debug Patroni when needed.
+
+Can I restart Postgres nodes directly?
+    No, you should **not** attempt to manage Postgres directly!
+
+    Any attempt of bouncing the Postgres server without Patroni can lead your cluster to face failovers.
+
+    If you need to manage the Postgres server, do that through the ways exposed by Patroni.
+
+Is Patroni able to take over management of an already existing Postgres cluster?
+    Yes, it can!
+
+    Please refer to :ref:`existing_data` for detailed instructions.
+
+How does Patroni manage Postgres?
+    Patroni takes care of bringing Postgres up and down by running the Postgres binaries, like ``pg_ctl`` and ``postgres``.
+
+    With that in mind you **MUST** disable any other sources that could manage the Postgres clusters, like the systemd units, e.g. ``postgresql.service``. Only Patroni should be able to start, stop and promote Postgres instances in the cluster. Not doing so may result in split-brain scenarios. For example: if the node running as a primary failed and the unit ``postgresql.service`` is enabled, it may bring Postgres back up and cause a split-brain.
+
+Concepts and requirements
+-------------------------
+
+Which are the applications that make part of Patroni?
+    Patroni basically ships a couple applications:
+
+    * ``patroni``: This is the Patroni agent, which takes care of managing a Postgres node;
+    * ``patronictl``: This is a command-line utility used to interact with a Patroni cluster (perform switchovers, restarts, changes in the configuration, etc.). Please find more information in :ref:`patronictl`.
+
+What is a ``standby cluster`` in Patroni?
+    It is a cluster that does not have any primary Postgres node running, i.e., there is no read/write member in the cluster.
+
+    These kinds of clusters exist to replicate data from another cluster and are usually useful when you want to replicate data across data centers.
+
+    There will be a leader in the cluster which will be a standby in charge of replicating changes from a remote Postgres node.
+    Then, there will be a set of standbys configured with cascading replication from such leader member.
+
+    **Note:** the standby cluster doesn't know anything about the source cluster which it is replicating from -- it can even use ``restore_command`` instead of WAL streaming, and may use an absolutely independent DCS cluster.
+
+    Refer to :ref:`standby_cluster` for more details.
+
+What is a ``leader`` in Patroni?
+    A ``leader`` in Patroni is like a coordinator of the cluster.
+
+    In a regular Patroni cluster, the ``leader`` will be the read/write node.
+
+    In a standby Patroni cluster, the ``leader`` (AKA ``standby leader``) will be in charge of replicating from a remote Postgres node, and cascading those changes to the other members of the standby cluster.
+
+Does Patroni require a minimum number of Postgres nodes in the cluster?
+    No, you can run Patroni with any number of Postgres nodes.
+
+    Remember: Patroni is decoupled from the DCS.
+
+What does ``pause`` mean in Patroni?
+    Pause is an operation exposed by Patroni so the user can ask Patroni to step back in regards to Postgres management.
+
+    That is mainly useful when you want to perform maintenance on the cluster, and would like to avoid that Patroni takes decisions related with HA, like failing over to a standby when you stop the primary.
+
+    You can find more information about that in :ref:`pause`.
+
+Automatic failover
+------------------
+
+How does the automatic failover mechanism of Patroni work?
+    Patroni automatic failover is based on what we call ``leader race``.
+
+    Patroni stores the cluster's status in the DCS, among them a ``leader`` lock which holds the name of the Patroni member which is the current ``leader`` of the cluster.
+
+    That ``leader`` lock has a time-to-live associated with it. If the leader node fails to update the lease of the ``leader`` lock in time, the key will eventually expire from the DCS.
+
+    When the ``leader`` lock expires, it triggers what Patroni calls a ``leader race``: all nodes start performing checks to determine if they are the best candidates for taking over the ``leader`` role.
+    Some of these checks include calls to the REST API of all other Patroni members.
+
+    All Patroni members that find themselves as the best candidate for taking over the ``leader`` lock will attempt to do so.
+    The first Patroni member that is able to take the ``leader`` lock will promote itself to a read/write node (or ``standby leader``), and the others will be configured to follow it.
+
+Can I temporarily disable automatic failover in the Patroni cluster?
+    Yes, you can!
+
+    You can achieve that by temporarily pausing the cluster.
+    This is typically useful for performing maintenance.
+
+    When you want to resume the automatic failover of the cluster, you just need to unpause it.
+
+    You can find more information about that in :ref:`pause`.
+
+Bootstrapping and standbys creation
+-----------------------------------
+
+How does Patroni create a primary Postgres node? What about a standby Postgres node?
+    By default Patroni will use ``initdb`` to bootstrap a fresh cluster, and ``pg_basebackup`` to create standby nodes from a copy of the ``leader`` member.
+
+    You can customize that behavior by writing your custom bootstrap methods, and your custom replica creation methods.
+
+    Custom methods are usually useful when you want to restore backups created by backup tools like pgBackRest or Barman, for example.
+
+    For detailed information please refer to :ref:`custom_bootstrap` and :ref:`custom_replica_creation`.
+
+Monitoring
+----------
+
+How can I monitor my Patroni cluster?
+    Patroni exposes a couple handy endpoints in its :ref:`rest_api`:
+
+    * ``/metrics``: exposes monitoring metrics in a format that can be consumed by Prometheus;
+    * ``/patroni``: exposes the status of the cluster in a JSON format. The information shown here is very similar to what is shown by the ``/metrics`` endpoint.
+
+    You can use those endpoints to implement monitoring checks.
diff --git a/docs/index.rst b/docs/index.rst
index f5a4f2d4..84d02db1 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -36,6 +36,7 @@ Currently supported PostgreSQL versions: 9.3 to 16.
    existing_data
    security
    ha_multi_dc
+   faq
    releases
    CONTRIBUTING
 
diff --git a/docs/patroni_configuration.rst b/docs/patroni_configuration.rst
index 1d6c0777..06d71426 100644
--- a/docs/patroni_configuration.rst
+++ b/docs/patroni_configuration.rst
@@ -30,6 +30,7 @@ There are 3 types of Patroni configuration:
 	It is possible to set/override some of the "Local" configuration parameters with environment variables.
 	Environment configuration is very useful when you are running in a dynamic environment and you don't know some of the parameters in advance (for example it's not possible to know your external IP address when you are running inside ``docker``).
 
+.. _important_configuration_rules:
 
 Important rules
 ---------------
@@ -90,6 +91,7 @@ The parameters would be applied in the following order (run-time are given the h
 
 This allows configuration for all the nodes (2), configuration for a specific node using ``ALTER SYSTEM`` (3) and ensures that parameters essential to the running of Patroni are enforced (4), as well as leaves room for configuration tools that manage `postgresql.conf` directly without involving Patroni (1).
 
+.. _shared_memory_gucs:
 
 PostgreSQL parameters that touch shared memory
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -156,6 +158,7 @@ Patroni provides command-line interfaces for a Patroni :ref:`local configuration
 - Create a Patroni configuration file for the locally running PostgreSQL instance (e.g. as a preparation step for the :ref:`Patroni integration <existing_data>`);
 - Validate a given Patroni configuration file.
 
+.. _generate_sample_config:
 
 Sample Patroni configuration
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -183,6 +186,7 @@ Parameters
 
 ``configfile`` - full path to the configuration file used to store the result. If not provided, the result is sent to ``stdout``.
 
+.. _generate_config:
 
 Patroni configuration for a running instance
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
diff --git a/docs/replica_bootstrap.rst b/docs/replica_bootstrap.rst
index 861a0e86..a6e3db8e 100644
--- a/docs/replica_bootstrap.rst
+++ b/docs/replica_bootstrap.rst
@@ -71,6 +71,22 @@ Makes the configured ``command`` to be called additionally with ``--arg1=value1
 
  .. note:: Bootstrap methods are neither chained, nor fallen-back to the default one in case the primary one fails
 
+As an example, you are able to bootstrap a fresh Patroni cluster from a Barman backup with a configuration like this:
+
+.. code:: YAML
+
+    bootstrap:
+        method: barman
+        barman:
+            keep_existing_recovery_conf: true
+            command: patroni_barman_recover
+            api-url: https://barman-host:7480
+            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``.
 
 .. _custom_replica_creation:
 
@@ -125,6 +141,25 @@ example: pgbackrest
         basebackup:
             max-rate: '100M'
 
+example: Barman
+
+.. code:: YAML
+
+    postgresql:
+        create_replica_methods:
+            - barman
+            - basebackup
+        barman:
+            command: patroni_barman_recover
+            api-url: https://barman-host:7480
+            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``.
 
 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
diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index 00272100..a03c05d2 100644
--- a/docs/rest_api.rst
+++ b/docs/rest_api.rst
@@ -426,6 +426,7 @@ Cluster status endpoints
       ]
     ]
 
+.. _config_endpoint:
 
 Config endpoint
 ---------------
@@ -666,6 +667,7 @@ There are a couple of checks that a member of a cluster should pass to be able t
 	- its lag exceeds the maximum replication lag allowed;
 	- it has the timeline number smaller than the last known cluster timeline.
 
+.. _restart_endpoint:
 
 Restart endpoint
 ----------------
@@ -682,6 +684,7 @@ Restart endpoint
 
 ``POST /restart`` and ``DELETE /restart`` endpoints are used by :ref:`patronictl_restart` and :ref:`patronictl flush cluster-name restart <patronictl_flush_parameters>` respectively.
 
+.. _reload_endpoint:
 
 Reload endpoint
 ---------------
diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst
index fc313fd0..a4a33134 100644
--- a/docs/yaml_configuration.rst
+++ b/docs/yaml_configuration.rst
@@ -133,6 +133,7 @@ ZooKeeper
 -  **key_password**: (optional) The client key password.
 -  **verify**: (optional) Whether to verify certificate or not. Defaults to ``true``.
 -  **set_acls**: (optional) If set, configure Kazoo to apply a default ACL to each ZNode that it creates. ACLs will assume 'x509' schema and should be specified as a dictionary with the principal as the key and one or more permissions as a list in the value.  Permissions may be one of ``CREATE``, ``READ``, ``WRITE``, ``DELETE`` or ``ADMIN``.  For example, ``set_acls: {CN=principal1: [CREATE, READ], CN=principal2: [ALL]}``.
+-  **auth_data**: (optional) Authentication credentials to use for the connection. Should be a dictionary in the form that `scheme` is the key and `credential` is the value. Defaults to empty dictionary.
 
 .. note::
     It is required to install ``kazoo>=2.6.0`` to support SSL.
diff --git a/patroni/config.py b/patroni/config.py
index ee0b2710..00dba9d9 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -791,7 +791,7 @@ class Config(object):
                               'SERVICE_TAGS', 'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL',
                               'POD_IP', 'PORTS', 'LABELS', 'BYPASS_API_SERVICE', 'RETRIABLE_HTTP_CODES', 'KEY_PASSWORD',
                               'USE_SSL', 'SET_ACLS', 'GROUP', 'DATABASE', 'LEADER_LABEL_VALUE', 'FOLLOWER_LABEL_VALUE',
-                              'STANDBY_LEADER_LABEL_VALUE', 'TMP_ROLE_LABEL') and name:
+                              'STANDBY_LEADER_LABEL_VALUE', 'TMP_ROLE_LABEL', 'AUTH_DATA') and name:
                     value = os.environ.pop(param)
                     if name == 'CITUS':
                         if suffix == 'GROUP':
@@ -802,7 +802,7 @@ class Config(object):
                         value = value and parse_int(value)
                     elif suffix in ('HOSTS', 'PORTS', 'CHECKS', 'SERVICE_TAGS', 'RETRIABLE_HTTP_CODES'):
                         value = value and _parse_list(value)
-                    elif suffix in ('LABELS', 'SET_ACLS'):
+                    elif suffix in ('LABELS', 'SET_ACLS', 'AUTH_DATA'):
                         value = _parse_dict(value)
                     elif suffix in ('USE_PROXIES', 'REGISTER_SERVICE', 'USE_ENDPOINTS', 'BYPASS_API_SERVICE', 'VERIFY'):
                         value = parse_bool(value)
diff --git a/patroni/ctl.py b/patroni/ctl.py
index 83810dd3..b4c57873 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -255,7 +255,8 @@ def load_config(path: str, dcs_url: Optional[str]) -> Dict[str, Any]:
     return config
 
 
-option_format = click.option('--format', '-f', 'fmt', help='Output format (pretty, tsv, json, yaml)', default='pretty')
+option_format = click.option('--format', '-f', 'fmt', help='Output format', default='pretty',
+                             type=click.Choice(['pretty', 'tsv', 'json', 'yaml', 'yml']))
 option_watchrefresh = click.option('-w', '--watch', type=float, help='Auto update the screen every X seconds')
 option_watch = click.option('-W', is_flag=True, help='Auto update the screen every 2 seconds')
 option_force = click.option('--force', is_flag=True, help='Do not ask for confirmation at any point')
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index 3093ba06..3704b579 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -115,7 +115,8 @@ class ZooKeeper(AbstractDCS):
         self._client = PatroniKazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']),
                                           timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
                                           sleep_func=time.sleep), command_retry=KazooRetry(max_delay=1, max_tries=-1,
-                                          deadline=config['retry_timeout'], sleep_func=time.sleep), **kwargs)
+                                          deadline=config['retry_timeout'], sleep_func=time.sleep),
+                                          auth_data=list(config.get('auth_data', {}).items()), **kwargs)
 
         self.__last_member_data: Optional[Dict[str, Any]] = None
 
diff --git a/patroni/scripts/barman_recover.py b/patroni/scripts/barman_recover.py
new file mode 100644
index 00000000..1cffe34d
--- /dev/null
+++ b/patroni/scripts/barman_recover.py
@@ -0,0 +1,468 @@
+#!/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()
diff --git a/patroni/validator.py b/patroni/validator.py
index 10bc7b2a..016f4b2e 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -1042,7 +1042,8 @@ schema = Schema({
             Optional("key"): str,
             Optional("key_password"): str,
             Optional("verify"): bool,
-            Optional("set_acls"): dict
+            Optional("set_acls"): dict,
+            Optional("auth_data"): dict,
         },
         "kubernetes": {
             "labels": {},
diff --git a/setup.py b/setup.py
index 4c1c25b3..5bd9c71f 100644
--- a/setup.py
+++ b/setup.py
@@ -54,7 +54,8 @@ CONSOLE_SCRIPTS = ['patroni = patroni.__main__:main',
                    'patronictl = patroni.ctl:ctl',
                    'patroni_raft_controller = patroni.raft_controller:main',
                    "patroni_wale_restore = patroni.scripts.wale_restore:main",
-                   "patroni_aws = patroni.scripts.aws:main"]
+                   "patroni_aws = patroni.scripts.aws:main",
+                   "patroni_barman_recover = patroni.scripts.barman_recover:main"]
 
 
 class _Command(Command):


Test patch:
diff --git a/tests/test_barman_recover.py b/tests/test_barman_recover.py
new file mode 100644
index 00000000..4334d6ae
--- /dev/null
+++ b/tests/test_barman_recover.py
@@ -0,0 +1,364 @@
+import logging
+from mock import MagicMock, Mock, call, 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([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.mock_calls[i].args
+            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(repr(call_args[4]), "KeyError('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([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.mock_calls[i].args
+            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(repr(call_args[4]), "KeyError('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([call("some_id")] * 21)
+
+        self.assertEqual(mock_log_info.call_count, 21)
+        mock_log_info.assert_has_calls([call("Created the recovery operation with ID %s", "some_id")]
+                                       + [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([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([call("some_id")] * 21)
+
+        self.assertEqual(mock_log_info.call_count, 21)
+        mock_log_info.assert_has_calls([call("Created the recovery operation with ID %s", "some_id")]
+                                       + [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([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)