Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

docs/README.rst

@@ -25,82 +25,7 @@ We report new releases information :ref:`here <releases>`.
 Technical Requirements/Installation
 -----------------------------------
 
-**Pre-requirements for Mac OS**
-
-To install requirements on a Mac, run the following:
-
-::
-
-    brew install postgresql etcd haproxy libyaml python
-
-.. _psycopg2_install_options:
-
-**Psycopg**
-
-Starting from `psycopg2-2.8 <http://initd.org/psycopg/articles/2019/04/04/psycopg-28-released/>`__ the binary version of psycopg2 will no longer be installed by default. Installing it from the source code requires C compiler and postgres+python dev packages.
-Since in the python world it is not possible to specify dependency as ``psycopg2 OR psycopg2-binary`` you will have to decide how to install it.
-
-There are a few options available:
-
-1. Use the package manager from your distro
-
-::
-
-    sudo apt-get install python3-psycopg2  # install psycopg2 module on Debian/Ubuntu
-    sudo yum install python3-psycopg2      # install psycopg2 on RedHat/Fedora/CentOS
-
-2. Install psycopg2 from the binary package
-
-::
-
-    pip install psycopg2-binary
-
-3. Install psycopg2 from source
-
-::
-
-    pip install psycopg2>=2.5.4
-
-4. Use psycopg 3.0 instead of psycopg2
-
-::
-
-    pip install psycopg[binary]>=3.0.0
-
-**General installation for pip**
-
-Patroni can be installed with pip:
-
-::
-
-    pip install patroni[dependencies]
-
-where dependencies can be either empty, or consist of one or more of the following:
-
-etcd or etcd3
-    `python-etcd` module in order to use Etcd as Distributed Configuration Store (DCS)
-consul
-    `python-consul` module in order to use Consul as DCS
-zookeeper
-    `kazoo` module in order to use Zookeeper as DCS
-exhibitor
-    `kazoo` module in order to use Exhibitor as DCS (same dependencies as for Zookeeper)
-kubernetes
-    `kubernetes` module in order to use Kubernetes as DCS in Patroni
-raft
-    `pysyncobj` module in order to use python Raft implementation as DCS
-aws
-    `boto3` in order to use AWS callbacks
-
-For example, the command in order to install Patroni together with dependencies for Etcd as a DCS and AWS callbacks is:
-
-::
-
-    pip install patroni[etcd,aws]
-
-Note that external tools to call in the replica creation or custom bootstrap scripts (i.e. WAL-E) should be installed
-independently of Patroni.
-
+Go :ref:`here <installation>` for guidance on installing and upgrading Patroni on various platforms.
 
 .. _running_configuring:
 

docs/index.rst

@@ -22,6 +22,7 @@ Currently supported PostgreSQL versions: 9.3 to 15.
    :caption: Contents:
 
    README
+   installation
    patroni_configuration
    rest_api
    replica_bootstrap

docs/installation.rst

@@ -0,0 +1,201 @@
+.. _installation:
+
+Installation
+============
+
+Pre-requirements for Mac OS
+---------------------------
+
+To install requirements on a Mac, run the following:
+
+.. code-block:: shell
+
+    brew install postgresql etcd haproxy libyaml python
+
+.. _psycopg2_install_options:
+
+Psycopg
+-------
+
+Starting from `psycopg2-2.8`_ the binary version of psycopg2 will no longer be installed by default. Installing it from
+the source code requires C compiler and postgres+python dev packages. Since in the python world it is not possible to
+specify dependency as ``psycopg2 OR psycopg2-binary`` you will have to decide how to install it.
+
+There are a few options available:
+
+1. Use the package manager from your distro
+
+.. code-block:: shell
+
+    sudo apt-get install python3-psycopg2  # install psycopg2 module on Debian/Ubuntu
+    sudo yum install python3-psycopg2      # install psycopg2 on RedHat/Fedora/CentOS
+
+2. Install psycopg2 from the binary package
+
+.. code-block:: shell
+
+    pip install psycopg2-binary
+
+3. Install psycopg2 from source
+
+.. code-block:: shell
+
+    pip install psycopg2>=2.5.4
+
+4. Use psycopg 3.0 instead of psycopg2
+
+.. code-block:: shell
+
+    pip install psycopg[binary]>=3.0.0
+
+General installation for pip
+----------------------------
+
+Patroni can be installed with pip:
+
+.. code-block:: shell
+
+    pip install patroni[dependencies]
+
+where ``dependencies`` can be either empty, or consist of one or more of the following:
+
+etcd or etcd3
+    `python-etcd` module in order to use Etcd as Distributed Configuration Store (DCS)
+consul
+    `python-consul` module in order to use Consul as DCS
+zookeeper
+    `kazoo` module in order to use Zookeeper as DCS
+exhibitor
+    `kazoo` module in order to use Exhibitor as DCS (same dependencies as for Zookeeper)
+kubernetes
+    `kubernetes` module in order to use Kubernetes as DCS in Patroni
+raft
+    `pysyncobj` module in order to use python Raft implementation as DCS
+aws
+    `boto3` in order to use AWS callbacks
+
+For example, the command in order to install Patroni together with dependencies for Etcd as a DCS and AWS callbacks is:
+
+.. code-block:: shell
+
+    pip install patroni[etcd,aws]
+
+Note that external tools to call in the replica creation or custom bootstrap scripts (i.e. WAL-E) should be installed
+independently of Patroni.
+
+.. _package_installation:
+
+Package installation on Linux
+-----------------------------
+
+Patroni packages may be available for your operating system, produced by the Postgres community for:
+
+* RHEL, RockyLinux, AlmaLinux;
+* Debian and Ubuntu;
+* SUSE Enterprise Linux.
+
+You can also find packages for direct dependencies of Patroni, like python modules that might not be available in
+the official operating system repositories.
+
+For more information see the `PGDG repository`_ documentation.
+
+If you are on a RedHat Enterprise Linux derivative operating system you may also require packages from EPEL, see
+`EPEL repository`_ documentation.
+
+Once you have installed the PGDG repository for your OS you can install patroni.
+
+.. note::
+
+    Patroni packages are not maintained by the Patroni developers, but rather by the Postgres community. If you
+    require support please first try connecting on `Postgres slack`_.
+
+Installing on Debian derivatives
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+With PGDG repo installed, see :ref:`above <package_installation>`, install Patroni via apt run:
+
+.. code-block:: shell
+
+    apt-get install patroni
+
+Installing on RedHat derivatives
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+With PGDG repo installed, see :ref:`above <package_installation>`, install patroni with an etcd DCS via dnf on RHEL 9
+(and derivatives) run:
+
+.. code-block:: shell
+
+    dnf install patroni patroni-etcd
+
+You can install etcd from PGDG if your RedHat derivative distribution does not provide packages. On the nodes that will
+host the DCS run:
+
+.. code-block:: shell
+
+    dnf install 'dnf-command(config-manager)'
+    dnf config-manager --enable pgdg-rhel9-extras
+    dnf install etcd
+
+You can replace the version of RHEL with `8` in the repo to make `pgdg-rhel8-extras` if needed. The repo name is still
+`pgdg-rhelN-extras` on RockyLinux, AlmaLinux, Oracle Linux, etc...
+
+Installing on SUSE Enterprise Linux
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+You might need to enable the SUSE PackageHub repositories for some dependencies. see `SUSE PackageHub`_ documentation.
+
+For SLES 15 with PGDG repo installed, see :ref:`above <package_installation>`, you can install patroni using:
+
+.. code-block:: shell
+
+    zypper install patroni patroni-etcd
+
+With the SUSE PackageHub repo enabled you can also install etcd:
+
+.. code-block:: shell
+
+    SUSEConnect -p PackageHub/15.5/x86_64
+    zypper install etcd
+
+Upgrading
+---------
+
+Upgrading patroni is a very simple process, just update the software installation and restart the Patroni daemon on
+each node in the cluster.
+
+However, restarting the Patroni daemon will result in a Postgres database restart. In some situations this may cause
+a failover of the primary node in your cluster, therefore it is recommended to put the cluster into maintenance mode
+until the Patroni daemon restart has been completed.
+
+To put the cluster in maintenance mode, run the following command on one of the patroni nodes:
+
+.. code-block:: shell
+
+    patronictl pause --wait
+
+Then on each node in the cluster, perform the package upgrade required for your OS:
+
+.. code-block:: shell
+
+    apt-get update && apt-get install patroni patroni-etcd
+
+Restart the patroni daemon process on each node:
+
+.. code-block:: shell
+
+    systemctl restart patroni
+
+Then finally resume monitoring of Postgres with patroni to take it out of maintenance mode:
+
+.. code-block:: shell
+
+    patronictl resume --wait
+
+The cluster will now be full operational with the new version of Patroni.
+
+.. _psycopg2-2.8: http://initd.org/psycopg/articles/2019/04/04/psycopg-28-released/
+.. _PGDG repository: https://www.postgresql.org/download/linux/
+.. _EPEL repository: https://docs.fedoraproject.org/en-US/epel/
+.. _SUSE PackageHub: https://packagehub.suse.com/how-to-use/
+.. _Postgres slack: http://pgtreats.info/slack-invite

docs/patroni_configuration.rst

@@ -90,6 +90,42 @@ 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).
 
 
+PostgreSQL parameters that touch shared memory
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+PostgreSQL has some parameters that determine the size of the shared memory used by them:
+
+- **max_connections**
+- **max_prepared_transactions**
+- **max_locks_per_transaction**
+- **max_wal_senders**
+- **max_worker_processes**
+
+Changing these parameters require a PostgreSQL restart to take effect, and their shared memory structures cannot be smaller on the standby nodes than on the primary node.
+
+As explained before, Patroni restrict changing their values through :ref:`dynamic configuration <dynamic_configuration>`, which usually consists of:
+
+1. Applying changes through ``patronictl edit-config`` (or via REST API ``/config`` endpoint)
+2. Restarting nodes through ``patronictl restart`` (or via REST API ``/restart`` endpoint)
+
+**Note:** please keep in mind that you should perform a restart of the PostgreSQL nodes through ``patronictl restart`` command, or via REST API ``/restart`` endpoint. An attempt to restart PostgreSQL by restarting the Patroni daemon, e.g. by executing ``systemctl restart patroni``, can cause a failover to occur in the cluster, if you are restarting the primary node.
+
+However, as those settings manage shared memory, some extra care should be taken when restarting the nodes:
+
+* If you want to **increase** the value of any of those settings:
+
+   1. Restart all standbys first
+   2. Restart the primary after that
+
+* If you want to **decrease** the value of any of those settings:
+
+   1. Restart the primary first
+   2. Restart all standbys after that
+
+**Note:** if you attempt to restart all nodes in one go after **decreasing** the value of any of those settings, Patroni will ignore the change and restart the standby with the original setting value, thus requiring that you restart the standbys again later. Patroni does that to prevent the standby to enter in an infinite crash loop, because PostgreSQL quits with a `FATAL` message if you attempt to set any of those parameters to a value lower than what is visible in ``pg_controldata`` on the Standby node. In other words, we can only decrease the setting on the standby once its ``pg_controldata`` is up-to-date with the primary in regards to these changes on the primary.
+
+More information about that can be found at `PostgreSQL Administrator's Overview <https://www.postgresql.org/docs/current/hot-standby.html#HOT-STANDBY-ADMIN>`__.
+
 Patroni configuration parameters
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

docs/pause.rst

@@ -19,7 +19,7 @@ When Patroni runs in a paused mode, it does not change the state of PostgreSQL,
 
 - For the Postgres primary with the leader lock Patroni updates the lock. If the node with the leader lock stops being the primary (i.e. is demoted manually), Patroni will release the lock instead of promoting the node back.
 
-- Manual unscheduled restart, reinitialize and manual failover are allowed. Manual failover is only allowed if the node to failover to is specified. In the paused mode, manual failover does not require a running primary node.
+- Manual unscheduled restart, manual unscheduled failover/switchover and reinitialize are allowed. No scheduled action is allowed. Manual switchover is only allowed if the node to switch over to is specified.
 
 - If 'parallel' primaries are detected by Patroni, it emits a warning, but does not demote the primary without the leader lock.
 

docs/rest_api.rst

@@ -567,39 +567,111 @@ The above call removes ``postgresql.parameters.max_connections`` from the dynami
 Switchover and failover endpoints
 ---------------------------------
 
-``POST /switchover`` or ``POST /failover``. These endpoints are very similar to each other. There are a couple of minor differences though:
+.. _switchover_api:
 
-1. The failover endpoint allows to perform a manual failover when there are no healthy nodes, but at the same time it will not allow you to schedule a switchover.
+Switchover
+^^^^^^^^^^
 
-2. The switchover endpoint is the opposite. It works only when the cluster is healthy (there is a leader) and allows to schedule a switchover at a given time.
+``/switchover`` endpoint only works when the cluster is healthy (there is a leader). It also allows to schedule a switchover at a given time.
 
+When calling ``/switchover`` endpoint a candidate can be specified but is not required, in contrast to ``/failover`` endpoint. If a candidate is not provided, all the eligible nodes of the cluster will participate in the leader race after the leader stepped down.
 
-In the JSON body of the ``POST`` request you must specify at least the ``leader`` or ``candidate`` fields and optionally the ``scheduled_at`` field if you want to schedule a switchover at a specific time.
+In the JSON body of the ``POST`` request you must specify the ``leader`` field. The ``candidate`` and the ``scheduled_at`` fields are optional and can be used to schedule a switchover at a specific time.
 
+Depending on the situation, requests might return different HTTP status codes and bodies. Status code **200** is returned when the switchover or failover successfully completed. If the switchover was successfully scheduled, Patroni will return HTTP status code **202**. In case something went wrong, the error status code (one of **400**, **412**, or **503**) will be returned with some details in the response body.
 
-Example: perform a failover to the specific node:
+``DELETE /switchover`` can be used to delete the currently scheduled switchover.
+
+**Example:** perform a switchover to any healthy standby
 
 .. code-block:: bash
 
-    $ curl -s http://localhost:8009/failover -XPOST -d '{"candidate":"postgresql1"}'
-    Successfully failed over to "postgresql1"
+	$ curl -s http://localhost:8008/switchover -XPOST -d '{"leader":"postgresql1"}'
+	Successfully switched over to "postgresql2"
 
 
-Example: schedule a switchover from the leader to any other healthy replica in the cluster at a specific time:
+**Example:** perform a switchover to a specific node
 
 .. code-block:: bash
 
-    $ curl -s http://localhost:8008/switchover -XPOST -d \
-	    '{"leader":"postgresql0","scheduled_at":"2019-09-24T12:00+00"}'
-    Switchover scheduled
+	$ curl -s http://localhost:8008/switchover -XPOST -d \
+		'{"leader":"postgresql1","candidate":"postgresql2"}'
+	Successfully switched over to "postgresql2"
 
 
-Depending on the situation the request might finish with a different HTTP status code and body. The status code **200** is returned when the switchover or failover successfully completed. If the switchover was successfully scheduled, Patroni will return HTTP status code **202**. In case something went wrong, the error status code (one of **400**, **412** or **503**) will be returned with some details in the response body. For more information please check the source code of ``patroni/api.py:do_POST_failover()`` method.
+**Example:** schedule a switchover from the leader to any other healthy standby in the cluster at a specific time.
 
-- ``DELETE /switchover``: delete the scheduled switchover
+.. code-block:: bash
 
-The ``POST /switchover`` and ``POST failover`` endpoints are used by ``patronictl switchover`` and ``patronictl failover``, respectively.
-The ``DELETE /switchover`` is used by ``patronictl flush <cluster-name> switchover``.
+	$ curl -s http://localhost:8008/switchover -XPOST -d \
+		'{"leader":"postgresql0","scheduled_at":"2019-09-24T12:00+00"}'
+	Switchover scheduled
+
+
+Failover
+^^^^^^^^
+
+``/failover`` endpoint can be used to perform a manual failover when there are no healthy nodes (e.g. to an asynchronous standby if all synchronous standbys are not healthy enough to promote). However there is no requirement for a cluster not to have leader - failover can also be run on a healthy cluster.
+
+In the JSON body of the ``POST`` request you must specify the ``candidate`` field. If the ``leader`` field is specified, a switchover is triggered instead.
+
+**Example:**
+
+.. code-block:: bash
+
+	$ curl -s http://localhost:8008/failover -XPOST -d '{"candidate":"postgresql1"}'
+	Successfully failed over to "postgresql1"
+
+.. warning::
+	:ref:`Be very careful <failover_healthcheck>` when using this endpoint, as this can cause data loss in certain situations. In most cases, :ref:`the switchover endpoint <switchover_api>` satisfies the administrator's needs. 
+
+
+``POST /switchover`` and ``POST /failover`` endpoints are used by ``patronictl switchover`` and ``patronictl failover``, respectively.
+
+``DELETE /switchover`` is used by ``patronictl flush <cluster-name> switchover``.
+
+.. list-table:: Failover/Switchover comparison
+   :widths: 25 25 25
+   :header-rows: 1
+
+   * -
+     - Failover
+     - Switchover
+   * - Requires leader specified
+     - no
+     - yes
+   * - Requires candidate specified
+     - yes
+     - no
+   * - Can be run in pause
+     - yes
+     - yes (only to a specific candidate)
+   * - Can be scheduled
+     - no
+     - yes (if not in pause)
+
+.. _failover_healthcheck:
+
+Healthy standby
+^^^^^^^^^^^^^^^
+
+There are a couple of checks that a member of a cluster should pass to be able to participate in the leader race during a switchover or to become a leader as a failover/switchover candidate:
+
+- be reachable via Patroni API;
+- not have ``nofailover`` tag set to ``true``;
+- have watchdog fully functional (if required by the configuration);
+- in case of a switchover in a healthy cluster or an automatic failover, not exceed maximum replication lag (``maximum_lag_on_failover`` :ref:`configuration parameter <dynamic_configuration>`);
+- in case of a switchover in a healthy cluster or an automatic failover, not have a timeline number smaller than the cluster timeline if ``check_timeline`` :ref:`configuration parameter <dynamic_configuration>` is set to ``true``;
+- in :ref:`synchronous mode <synchronous_mode>`:
+
+  - In case of a switchover (both with and without a candidate): be listed in the ``/sync`` key members;
+  - For a failover in both healthy and unhealthy clusters, this check is omitted.
+
+.. warning::
+    In case of a manual failover in a cluster without a leader, a candidate will be allowed to promote even if:
+	- it is not in the ``/sync`` key members when synchronous mode is enabled;
+	- its lag exceeds the maximum replication lag allowed;
+	- it has the timeline number smaller than the last known cluster timeline.
 
 
 Restart endpoint

patroni/api.py

@@ -1095,7 +1095,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
             * ``412``: if operation is not possible;
             * ``503``: if unable to register the operation to the DCS;
             * HTTP status returned by :func:`parse_schedule`, if any error was observed while parsing the schedule;
-            * HTTP status returned by :func:`poll_failover_result` if the operation has been processed immediately.
+            * HTTP status returned by :func:`poll_failover_result` if the operation has been processed immediately;
+            * ``400``: if none of the above applies.
 
         .. note::
             If unable to parse the request body, then the request is silently discarded.
@@ -1122,15 +1123,22 @@ class RestApiHandler(BaseHTTPRequestHandler):
             data = 'Switchover could be performed only from a specific leader'
 
         if not data and scheduled_at:
-            if not leader:
-                data = 'Scheduled {0} is possible only from a specific leader'.format(action)
-            if not data and global_config.is_paused:
-                data = "Can't schedule {0} in the paused state".format(action)
-            if not data:
+            if action == 'failover':
+                data = "Failover can't be scheduled"
+            elif global_config.is_paused:
+                data = "Can't schedule switchover in the paused state"
+            else:
                 (status_code, data, scheduled_at) = self.parse_schedule(scheduled_at, action)
 
         if not data and global_config.is_paused and not candidate:
-            data = action.title() + ' is possible only to a specific candidate in a paused state'
+            data = 'Switchover is possible only to a specific candidate in a paused state'
+
+        if action == 'failover' and leader:
+            logger.warning('received failover request with leader specifed - performing switchover instead')
+            action = 'switchover'
+
+        if not data and leader == candidate:
+            data = 'Switchover target and source are the same'
 
         if not data and not scheduled_at:
             data = self.is_failover_possible(cluster, leader, candidate, action)
@@ -1147,7 +1155,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     status_code, data = self.poll_failover_result(cluster.leader and cluster.leader.name,
                                                                   candidate, action)
             else:
-                data = 'failed to write {0} key into DCS'.format(action)
+                data = 'failed to write failover key into DCS'
                 status_code = 503
         # pyright thinks ``status_code`` can be ``None`` because ``parse_schedule`` call may return ``None``. However,
         # if that's the case, ``status_code`` will be overwritten somewhere between ``parse_schedule`` and

patroni/ctl.py

@@ -46,6 +46,7 @@ try:
 except ImportError:  # pragma: no cover
     from cdiff import markup_to_pager, PatchStream  # pyright: ignore [reportMissingModuleSource]
 
+from .config import Config, get_global_config
 from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member
 from .exceptions import PatroniException
 from .postgresql.misc import postgres_version_to_int
@@ -225,8 +226,6 @@ def load_config(path: str, dcs_url: Optional[str]) -> Dict[str, Any]:
     :raises:
         :class:`PatroniCtlException`: if *path* does not exist or is not readable.
     """
-    from patroni.config import Config
-
     if not (os.path.exists(path) and os.access(path, os.R_OK)):
         if path != CONFIG_FILE_PATH:    # bail if non-default config location specified but file not found / readable
             raise PatroniCtlException('Provided config file {0} not existing or no read rights.'
@@ -1013,7 +1012,6 @@ def reload(obj: Dict[str, Any], cluster_name: str, member_names: List[str],
         if r.status == 200:
             click.echo('No changes to apply on member {0}'.format(member.name))
         elif r.status == 202:
-            from patroni.config import get_global_config
             config = get_global_config(cluster)
             click.echo('Reload request received for member {0} and will be processed within {1} seconds'.format(
                 member.name, config.get('loop_wait') or dcs.loop_wait)
@@ -1095,7 +1093,6 @@ def restart(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member
         content['postgres_version'] = version
 
     if scheduled_at:
-        from patroni.config import get_global_config
         if get_global_config(cluster).is_paused:
             raise PatroniCtlException("Can't schedule restart in the paused state")
         content['schedule'] = scheduled_at.isoformat()
@@ -1223,19 +1220,22 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
             dcs = get_dcs(obj, cluster_name, group)
             cluster = dcs.get_cluster()
 
-    if action == 'switchover' and (cluster.leader is None or not cluster.leader.name):
-        raise PatroniCtlException('This cluster has no leader')
+    global_config = get_global_config(cluster)
 
-    if leader is None:
-        if force or action == 'failover':
-            leader = cluster.leader and cluster.leader.name
-        else:
-            from patroni.config import get_global_config
-            prompt = 'Standby Leader' if get_global_config(cluster).is_standby_cluster else 'Primary'
-            leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.member.name))
+    # leader has to be be defined for switchover only
+    if action == 'switchover':
+        if cluster.leader is None or not cluster.leader.name:
+            raise PatroniCtlException('This cluster has no leader')
 
-    if leader is not None and cluster.leader and cluster.leader.member.name != leader:
-        raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.format(leader, cluster_name))
+        if leader is None:
+            if force:
+                leader = cluster.leader.name
+            else:
+                prompt = 'Standby Leader' if global_config.is_standby_cluster else 'Primary'
+                leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name))
+
+        if cluster.leader.name != leader:
+            raise PatroniCtlException(f'Member {leader} is not the leader of cluster {cluster_name}')
 
     # excluding members with nofailover tag
     candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover]
@@ -1255,7 +1255,16 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
         raise PatroniCtlException(action.title() + ' target and source are the same.')
 
     if candidate and candidate not in candidate_names:
-        raise PatroniCtlException('Member {0} does not exist in cluster {1}'.format(candidate, cluster_name))
+        raise PatroniCtlException(
+            f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover')
+
+    if all((not force,
+            action == 'failover',
+            global_config.is_synchronous_mode,
+            not cluster.sync.is_empty,
+            not cluster.sync.matches(candidate, True))):
+        if click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}'):
+            raise PatroniCtlException('Aborting ' + action)
 
     scheduled_at_str = None
     scheduled_at = None
@@ -1268,25 +1277,29 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
 
         scheduled_at = parse_scheduled(scheduled)
         if scheduled_at:
-            from patroni.config import get_global_config
-            if get_global_config(cluster).is_paused:
+            if global_config.is_paused:
                 raise PatroniCtlException("Can't schedule switchover in the paused state")
             scheduled_at_str = scheduled_at.isoformat()
 
-    failover_value = {'leader': leader, 'candidate': candidate, 'scheduled_at': scheduled_at_str}
+    failover_value = {'candidate': candidate}
+    if action == 'switchover':
+        failover_value['leader'] = leader
+    if scheduled_at_str:
+        failover_value['scheduled_at'] = scheduled_at_str
 
     logging.debug(failover_value)
 
     # By now we have established that the leader exists and the candidate exists
     if not force:
-        demote_msg = ', demoting current leader ' + leader if leader else ''
+        demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else ''
         if scheduled_at_str:
-            if not click.confirm('Are you sure you want to schedule {0} of cluster {1} at {2}{3}?'
-                                 .format(action, cluster_name, scheduled_at_str, demote_msg)):
+            # only switchover can be scheduled
+            if not click.confirm(f'Are you sure you want to schedule switchover of cluster '
+                                 f'{cluster_name} at {scheduled_at_str}{demote_msg}?'):
+                # action as a var to catch a regression in the tests
                 raise PatroniCtlException('Aborting scheduled ' + action)
         else:
-            if not click.confirm('Are you sure you want to {0} cluster {1}{2}?'
-                                 .format(action, cluster_name, demote_msg)):
+            if not click.confirm(f'Are you sure you want to {action} cluster {cluster_name}{demote_msg}?'):
                 raise PatroniCtlException('Aborting ' + action)
 
     r = None
@@ -1332,6 +1345,8 @@ def failover(obj: Dict[str, Any], cluster_name: str, group: Optional[int],
 
     .. note::
         If *leader* is given perform a switchover instead of a failover.
+        This behavior is deprecated. ``--leader`` option support will be
+        removed in the next major release.
 
     .. seealso::
         Refer to :func:`_do_failover_or_switchover` for details.
@@ -1345,7 +1360,12 @@ def failover(obj: Dict[str, Any], cluster_name: str, group: Optional[int],
     :param candidate: name of a standby member to be promoted. Nodes that are tagged with ``nofailover`` cannot be used.
     :param force: perform the failover or switchover without asking for confirmations.
     """
-    action = 'switchover' if leader else 'failover'
+    action = 'failover'
+    if leader:
+        action = 'switchover'
+        click.echo(click.style(
+            'Supplying a leader name using this command is deprecated and will be removed in a future version of'
+            ' Patroni, change your scripts to use `switchover` instead.\nExecuting switchover!', fg='red'))
     _do_failover_or_switchover(obj, action, cluster_name, group, leader, candidate, force)
 
 
@@ -1718,7 +1738,6 @@ def wait_until_pause_is_applied(dcs: AbstractDCS, paused: bool, old_cluster: Clu
     :param old_cluster: original cluster information before pause or unpause has been requested. Used to report which
         nodes are still pending to have ``pause`` equal *paused* at a given point in time.
     """
-    from patroni.config import get_global_config
     config = get_global_config(old_cluster)
 
     click.echo("'{0}' request sent, waiting until it is recognized by all nodes".format(paused and 'pause' or 'resume'))
@@ -1756,7 +1775,6 @@ def toggle_pause(config: Dict[str, Any], cluster_name: str, group: Optional[int]
             * ``pause`` state is already *paused*; or
             * cluster contains no accessible members.
     """
-    from patroni.config import get_global_config
     dcs = get_dcs(config, cluster_name, group)
     cluster = dcs.get_cluster()
     if get_global_config(cluster).is_paused == paused:

patroni/ha.py

@@ -236,6 +236,17 @@ class Ha(object):
         """
         return self.is_quorum_commit_mode() and not self.cluster.sync.is_empty
 
+    def _get_failover_action_name(self) -> str:
+        """Return the currently requested manual failover action name or the default ``failover``.
+
+        :returns: :class:`str` representing the manually requested action (``manual failover`` if no leader
+            is specified in the ``/failover`` in DCS, ``switchover`` otherwise) or ``failover`` if
+            ``/failover`` is empty.
+        """
+        if not self.cluster.failover:
+            return 'failover'
+        return 'switchover' if self.cluster.failover.leader else 'manual failover'
+
     def load_cluster_from_dcs(self) -> None:
         cluster = self.dcs.get_cluster()
 
@@ -1143,11 +1154,12 @@ class Ha(object):
         """
         candidates = self.get_failover_candidates(exclude_failover_candidate)
 
+        action = self._get_failover_action_name()
         if self.is_synchronous_mode() and self.cluster.failover and self.cluster.failover.candidate and not candidates:
-            logger.warning('Failover candidate=%s does not match with sync_standbys=%s',
-                           self.cluster.failover.candidate, self.cluster.sync.sync_standby)
+            logger.warning('%s candidate=%s does not match with sync_standbys=%s',
+                           action.title(), self.cluster.failover.candidate, self.cluster.sync.sync_standby)
         elif not candidates:
-            logger.warning('manual failover: candidates list is empty')
+            logger.warning('%s: candidates list is empty', action)
 
         ret = False
         cluster_timeline = self.cluster.timeline
@@ -1174,15 +1186,18 @@ class Ha(object):
         failover = self.cluster.failover
         if TYPE_CHECKING:  # pragma: no cover
             assert failover is not None
-        if failover.candidate:  # manual failover to specific member
-            if failover.candidate == self.state_handler.name:  # manual failover to me
+
+        action = self._get_failover_action_name()
+
+        if failover.candidate:  # manual failover/switchover to specific member
+            if failover.candidate == self.state_handler.name:  # manual failover/switchover to me
                 return True
             elif self.is_paused():
                 # Remove failover key if the node to failover has terminated to avoid waiting for it indefinitely
                 # In order to avoid attempts to delete this key from all nodes only the primary is allowed to do it.
                 if not self.cluster.get_member(failover.candidate, fallback_to_leader=False)\
                         and self.state_handler.is_primary():
-                    logger.warning("manual failover: removing failover key because failover candidate is not running")
+                    logger.warning("%s: removing failover key because failover candidate is not running", action)
                     self.dcs.manual_failover('', '', version=failover.version)
                     return None
                 return False
@@ -1199,17 +1214,17 @@ class Ha(object):
                 st = self.fetch_node_status(member)
                 not_allowed_reason = st.failover_limitation()
                 if not_allowed_reason is None:  # node is healthy
-                    logger.info('manual failover: to %s, i am %s', st.member.name, self.state_handler.name)
+                    logger.info('%s: to %s, i am %s', action, st.member.name, self.state_handler.name)
                     return False
-                # we wanted to failover to specific member but it is not healthy
-                logger.warning('manual failover: member %s is %s', st.member.name, not_allowed_reason)
+                # we wanted to failover/switchover to specific member but it is not healthy
+                logger.warning('%s: member %s is %s', action, st.member.name, not_allowed_reason)
 
-            # at this point we should consider all members as a candidates for failover
+            # at this point we should consider all members as a candidates for failover/switchover
             # i.e. we assume that failover.candidate is None
         elif self.is_paused():
             return False
 
-        # try to pick some other members to failover and check that they are healthy
+        # try to pick some other members for switchover and check that they are healthy
         if failover.leader:
             if self.state_handler.name == failover.leader:  # I was the leader
                 # exclude desired member which is unhealthy if it was specified
@@ -1267,8 +1282,8 @@ class Ha(object):
 
         if self.cluster.failover:
             # When doing a switchover in synchronous mode only synchronous nodes and former leader are allowed to race
-            if self.sync_mode_is_active() and not self.cluster.sync.matches(self.state_handler.name, True) and \
-               self.cluster.failover.leader:
+            if self.cluster.failover.leader and self.sync_mode_is_active() \
+                    and not self.cluster.sync.matches(self.state_handler.name, True):
                 return False
             return self.manual_failover_process_no_leader() or False
 
@@ -1434,28 +1449,35 @@ class Ha(object):
 
         :returns: action message if demote was initiated, None if no action was taken"""
         failover = self.cluster.failover
+        # if there is no failover key or
+        # I am holding the lock but am not primary = I am the standby leader,
+        # then do nothing
         if not failover or (self.is_paused() and not self.state_handler.is_primary()):
             return
 
+        action = self._get_failover_action_name()
+        bare_action = action.replace('manual ', '')
+
+        # it is not the time for the scheduled switchover yet, do nothing
         if (failover.scheduled_at and not
-            self.should_run_scheduled_action("failover", failover.scheduled_at, lambda:
+            self.should_run_scheduled_action(bare_action, failover.scheduled_at, lambda:
                                              self.dcs.manual_failover('', '', version=failover.version))):
             return
 
         if not failover.leader or failover.leader == self.state_handler.name:
             if not failover.candidate or failover.candidate != self.state_handler.name:
                 if not failover.candidate and self.is_paused():
-                    logger.warning('Failover is possible only to a specific candidate in a paused state')
+                    logger.warning('%s is possible only to a specific candidate in a paused state', action.title())
                 elif self.is_failover_possible():
-                    ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',))
-                    return ret or 'manual failover: demoting myself'
+                    ret = self._async_executor.try_run_async(f'{action}: demote', self.demote, ('graceful',))
+                    return ret or f'{action}: demoting myself'
                 else:
-                    logger.warning('manual failover: no healthy members found, failover is not possible')
+                    logger.warning('%s: no healthy members found, %s is not possible',
+                                   action, bare_action)
             else:
-                logger.warning('manual failover: I am already the leader, no need to failover')
+                logger.warning('%s: I am already the leader, no need to %s', action, bare_action)
         else:
-            logger.warning('manual failover: leader name does not match: %s != %s',
-                           failover.leader, self.state_handler.name)
+            logger.warning('%s: leader name does not match: %s != %s', action, failover.leader, self.state_handler.name)
 
         logger.info('Cleaning up failover key')
         self.dcs.manual_failover('', '', version=failover.version)
@@ -1512,6 +1534,7 @@ class Ha(object):
                     self._delete_leader()
                     return 'removed leader lock because postgres is not running as primary'
 
+            # update lock to avoid split-brain
             if self.update_lock(True):
                 msg = self.process_manual_failover_from_leader()
                 if msg is not None:
@@ -2182,9 +2205,10 @@ class Ha(object):
         def is_eligible(node: Member) -> bool:
             # If quorum commit is requested we want to check all nodes (even not voters),
             # because they could get enough votes and reach necessary quorum + 1.
-            # TODO: allow manual failover (=no leader specified) to async node
-            if self.sync_mode_is_active() \
-                    and not (self.is_quorum_commit_mode() or self.cluster.sync.matches(node.name)):
+            # in synchronous mode we allow failover (not switchover!) to async node
+            if self.sync_mode_is_active()\
+                    and not (self.is_quorum_commit_mode() or self.cluster.sync.matches(node.name))\
+                    and not (failover and not failover.leader):
                 return False
             # Don't spend time on "nofailover" nodes checking.
             # We also don't need nodes which we can't query with the api in the list.

tests/test_api.py

@@ -519,86 +519,161 @@ class TestRestApiHandler(unittest.TestCase):
 
         post = 'POST /switchover HTTP/1.0' + self._authorization + '\nContent-Length: '
 
-        MockRestApiServer(RestApiHandler, post + '7\n\n{"1":2}')
+        # Invalid content
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            MockRestApiServer(RestApiHandler, post + '7\n\n{"1":2}')
+            response_mock.assert_called_with(400, 'Switchover could be performed only from a specific leader')
 
+        # Empty content
         request = post + '0\n\n'
         MockRestApiServer(RestApiHandler, request)
 
-        cluster.leader.name = 'postgresql1'
-        MockRestApiServer(RestApiHandler, request)
+        # [Switchover without a candidate]
 
-        request = post + '25\n\n{"leader": "postgresql1"}'
-
-        with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)):
+        # Cluster with only a leader
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            cluster.leader.name = 'postgresql1'
+            request = post + '25\n\n{"leader": "postgresql1"}'
             MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(
+                412, 'switchover is not possible: cluster does not have members except leader')
 
-        for is_synchronous_mode in (True, False):
-            with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)):
+        # Switchover in pause mode
+        with patch.object(RestApiHandler, 'write_response') as response_mock, \
+             patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)):
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(
+                400, 'Switchover is possible only to a specific candidate in a paused state')
+
+        # No healthy nodes to promote in both sync and async mode
+        for is_synchronous_mode, response in (
+                (True, 'switchover is not possible: can not find sync_standby'),
+                (False, 'switchover is not possible: cluster does not have members except leader')):
+            with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)), \
+                 patch.object(RestApiHandler, 'write_response') as response_mock:
                 MockRestApiServer(RestApiHandler, request)
+                response_mock.assert_called_with(412, response)
 
-        cluster.leader.name = 'postgresql2'
-        request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
-        MockRestApiServer(RestApiHandler, request)
+        # [Switchover to the candidate specified]
 
+        # Candidate to promote is the same as the leader specified
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            request = post + '53\n\n{"leader": "postgresql2", "candidate": "postgresql2"}'
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(400, 'Switchover target and source are the same')
+
+        # Current leader is different from the one specified
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            cluster.leader.name = 'postgresql2'
+            request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(412, 'leader name does not match')
+
+        # Candidate to promote is not a member of the cluster
         cluster.leader.name = 'postgresql1'
         cluster.sync.matches.return_value = False
-        for is_synchronous_mode in (True, False):
-            with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)):
+        for is_synchronous_mode, response in (
+                (True, 'candidate name does not match with sync_standby'), (False, 'candidate does not exists')):
+            with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)), \
+                 patch.object(RestApiHandler, 'write_response') as response_mock:
                 MockRestApiServer(RestApiHandler, request)
+                response_mock.assert_called_with(412, response)
 
         cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}),
                            Member(0, 'postgresql2', 30, {'api_url': 'http'})]
-        MockRestApiServer(RestApiHandler, request)
 
-        cluster.failover = None
-        MockRestApiServer(RestApiHandler, request)
+        # Failover key is empty in DCS
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            cluster.failover = None
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(503, 'Switchover failed')
 
-        dcs.get_cluster.side_effect = [cluster]
-        MockRestApiServer(RestApiHandler, request)
+        # Result polling failed
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            dcs.get_cluster.side_effect = [cluster]
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(503, 'Switchover status unknown')
 
-        cluster2 = cluster.copy()
-        cluster2.leader.name = 'postgresql0'
-        cluster2.is_unlocked.return_value = False
-        dcs.get_cluster.side_effect = [cluster, cluster2]
-        MockRestApiServer(RestApiHandler, request)
+        # Switchover to a node different from the candidate specified
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            cluster2 = cluster.copy()
+            cluster2.leader.name = 'postgresql0'
+            cluster2.is_unlocked.return_value = False
+            dcs.get_cluster.side_effect = [cluster, cluster2]
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(200, 'Switched over to "postgresql0" instead of "postgresql2"')
 
-        cluster2.leader.name = 'postgresql2'
-        dcs.get_cluster.side_effect = [cluster, cluster2]
-        MockRestApiServer(RestApiHandler, request)
+        # Successful switchover to the candidate
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            cluster2.leader.name = 'postgresql2'
+            dcs.get_cluster.side_effect = [cluster, cluster2]
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(200, 'Successfully switched over to "postgresql2"')
+
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            dcs.manual_failover.return_value = False
+            dcs.get_cluster.side_effect = None
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(503, 'failed to write failover key into DCS')
 
-        dcs.get_cluster.side_effect = None
-        dcs.manual_failover.return_value = False
-        MockRestApiServer(RestApiHandler, request)
         dcs.manual_failover.return_value = True
 
-        with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])):
+        # Candidate is not healthy to be promoted
+        with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])), \
+             patch.object(RestApiHandler, 'write_response') as response_mock:
             MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(412, 'switchover is not possible: no good candidates have been found')
+
+        # [Scheduled switchover]
 
         # Valid future date
-        request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2",' +\
-                         ' "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
-        MockRestApiServer(RestApiHandler, request)
-        with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)), \
-                patch.object(MockPatroni, 'dcs') as d:
-            d.manual_failover.return_value = False
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2",' + \
+                             ' "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
             MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(202, 'Switchover scheduled')
 
-        # Exception: No timezone specified
-        request = post + '97\n\n{"leader": "postgresql1", "member": "postgresql2",' +\
-                         ' "scheduled_at": "6016-02-15T18:13:30.568224"}'
-        MockRestApiServer(RestApiHandler, request)
+        # Schedule in paused mode
+        with patch.object(RestApiHandler, 'write_response') as response_mock, \
+             patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)):
+            dcs.manual_failover.return_value = False
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(400, "Can't schedule switchover in the paused state")
+
+        # No timezone specified
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            request = post + '97\n\n{"leader": "postgresql1", "member": "postgresql2",' + \
+                             ' "scheduled_at": "6016-02-15T18:13:30.568224"}'
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(400, 'Timezone information is mandatory for the scheduled switchover')
 
-        # Exception: Scheduled in the past
         request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2", "scheduled_at": "'
-        MockRestApiServer(RestApiHandler, request + '1016-02-15T18:13:30.568224+01:00"}')
+
+        # Scheduled in the past
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            MockRestApiServer(RestApiHandler, request + '1016-02-15T18:13:30.568224+01:00"}')
+            response_mock.assert_called_with(422, 'Cannot schedule switchover in the past')
 
         # Invalid date
-        self.assertIsNotNone(MockRestApiServer(RestApiHandler, request + '2010-02-29T18:13:30.568224+01:00"}'))
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            MockRestApiServer(RestApiHandler, request + '2010-02-29T18:13:30.568224+01:00"}')
+            response_mock.assert_called_with(
+                422, 'Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601')
 
     def test_do_POST_failover(self):
         post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: '
-        MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}')
-        MockRestApiServer(RestApiHandler, post + '37\n\n{"candidate":"2","scheduled_at": "1"}')
+
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}')
+            response_mock.assert_called_once_with(400, 'Failover could be performed only to a specific candidate')
+
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            MockRestApiServer(RestApiHandler, post + '37\n\n{"candidate":"2","scheduled_at": "1"}')
+            response_mock.assert_called_once_with(400, "Failover can't be scheduled")
+
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            MockRestApiServer(RestApiHandler, post + '30\n\n{"leader":"1","candidate":"2"}')
+            response_mock.assert_called_once_with(412, 'leader name does not match')
 
     @patch.object(MockHa, 'is_leader', Mock(return_value=True))
     def test_do_POST_citus(self):

tests/test_ctl.py

@@ -21,10 +21,17 @@ from .test_ha import get_cluster_initialized_without_leader, get_cluster_initial
     get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader, get_cluster, Member
 
 
-@patch('patroni.ctl.load_config', Mock(return_value={
-    'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'}, 'ctl': {'certfile': 'a'},
-    'etcd': {'host': 'localhost:2379'}, 'citus': {'database': 'citus', 'group': 0},
-    'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5}}))
+DEFAULT_CONFIG = {
+    'scope': 'alpha',
+    'restapi': {'listen': '::', 'certfile': 'a'},
+    'ctl': {'certfile': 'a'},
+    'etcd': {'host': 'localhost:2379'},
+    'citus': {'database': 'citus', 'group': 0},
+    'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5}
+}
+
+
+@patch('patroni.ctl.load_config', Mock(return_value=DEFAULT_CONFIG))
 class TestCtl(unittest.TestCase):
     TEST_ROLES = ('master', 'primary', 'leader')
 
@@ -96,91 +103,131 @@ class TestCtl(unittest.TestCase):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
         mock_get_dcs.return_value.set_failover_value = Mock()
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
-        assert 'leader' in result.output
 
+        # Confirm
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
+        self.assertEqual(result.exit_code, 0)
+
+        # Abort
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\nN')
+        self.assertEqual(result.exit_code, 1)
+
+        # Without a candidate with --force option
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
+        self.assertEqual(result.exit_code, 0)
+
+        # Scheduled (confirm)
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
                                     input='leader\nother\n2300-01-01T12:23:00\ny')
-        assert result.exit_code == 0
+        self.assertEqual(result.exit_code, 0)
 
+        # Scheduled (abort)
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
+                                    '--scheduled', '2015-01-01T12:00:00+01:00'], input='leader\nother\n\nN')
+        self.assertEqual(result.exit_code, 1)
+
+        # Scheduled with --force option
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
+                                    '--force', '--scheduled', '2015-01-01T12:00:00+01:00'])
+        self.assertEqual(result.exit_code, 0)
+
+        # Scheduled in pause mode
         with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)):
             result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
                                               '--force', '--scheduled', '2015-01-01T12:00:00'])
-            assert result.exit_code == 1
-
-        # Aborting switchover, as we answer NO to the confirmation
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\nN')
-        assert result.exit_code == 1
-
-        # Aborting scheduled switchover, as we answer NO to the confirmation
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
-                                    '--scheduled', '2015-01-01T12:00:00+01:00'], input='leader\nother\n\nN')
-        assert result.exit_code == 1
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn("Can't schedule switchover in the paused state", result.output)
 
         # Target and source are equal
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
-        assert result.exit_code == 1
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Switchover target and source are the same', result.output)
 
-        # Reality is not part of this cluster
+        # Candidate is not a member of the cluster
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
-        assert result.exit_code == 1
-
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
-        assert 'Member' in result.output
-
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
-                                          '--force', '--scheduled', '2015-01-01T12:00:00+01:00'])
-        assert result.exit_code == 0
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Member Reality does not exist in cluster dummy or is tagged as nofailover', result.output)
 
         # Invalid timestamp
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force', '--scheduled', 'invalid'])
-        assert result.exit_code != 0
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Unable to parse scheduled timestamp', result.output)
 
         # Invalid timestamp
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
                                           '--force', '--scheduled', '2115-02-30T12:00:00+01:00'])
-        assert result.exit_code != 0
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Unable to parse scheduled timestamp', result.output)
 
         # Specifying wrong leader
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='dummy')
-        assert result.exit_code == 1
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Member dummy is not the leader of cluster dummy', result.output)
 
+        # Errors while sending Patroni REST API request
         with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
-            # Non-responding patroni
             result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
                                         input='leader\nother\n2300-01-01T12:23:00\ny')
-            assert 'falling back to DCS' in result.output
+            self.assertIn('falling back to DCS', result.output)
 
-        with patch.object(PoolManager, 'request') as mocked:
-            mocked.return_value.status = 500
+        with patch.object(PoolManager, 'request') as mock_api_request:
+            mock_api_request.return_value.status = 500
             result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
-            assert 'Switchover failed' in result.output
+            self.assertIn('Switchover failed', result.output)
 
-            mocked.return_value.status = 501
-            mocked.return_value.data = b'Server does not support this operation'
+            mock_api_request.return_value.status = 501
+            mock_api_request.return_value.data = b'Server does not support this operation'
             result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
-            assert 'Switchover failed' in result.output
+            self.assertIn('Switchover failed', result.output)
 
         # No members available
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_only_leader
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
-        assert result.exit_code == 1
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('No candidates found to switchover to', result.output)
 
         # No leader available
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_without_leader
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
-        assert result.exit_code == 1
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('This cluster has no leader', result.output)
+
+        # Citus cluster, no group number specified
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--force'], input='\n')
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('For Citus clusters the --group must me specified', result.output)
 
     @patch('patroni.ctl.get_dcs')
     @patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
+    @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse()))
     def test_failover(self, mock_get_dcs):
-        mock_get_dcs.return_value = self.e
-        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
         mock_get_dcs.return_value.set_failover_value = Mock()
-        result = self.runner.invoke(ctl, ['failover', 'dummy', '--force'], input='\n')
-        assert 'For Citus clusters the --group must me specified' in result.output
+
+        # No candidate specified
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
         result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n')
-        assert 'Failover could be performed only to a specific candidate' in result.output
+        self.assertIn('Failover could be performed only to a specific candidate', result.output)
+
+        cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
+
+        # Temp test to check a fallback to switchover if leader is specified
+        with patch('patroni.ctl._do_failover_or_switchover') as failover_func_mock:
+            result = self.runner.invoke(ctl, ['failover', '--leader', 'leader', 'dummy'], input='0\n')
+            self.assertIn('Supplying a leader name using this command is deprecated', result.output)
+            failover_func_mock.assert_called_once_with(
+                DEFAULT_CONFIG, 'switchover', 'dummy', None, 'leader', None, False)
+
+        # Failover to an async member in sync mode (confirm)
+        cluster.members.append(Member(0, 'async', 28, {'api_url': 'http://127.0.0.1:8012/patroni'}))
+        cluster.config.data['synchronous_mode'] = True
+        mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster)
+        result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny')
+        self.assertIn('Are you sure you want to failover to the asynchronous node async', result.output)
+
+        # Failover to an async member in sync mode (abort)
+        mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster)
+        result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N')
+        self.assertEqual(result.exit_code, 1)
 
     @patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd']))
     def test_get_dcs(self):

tests/test_ha.py

@@ -437,6 +437,7 @@ class TestHa(PostgresInit):
 
     def test_promote_without_watchdog(self):
         self.ha.has_lock = true
+        self.p.is_primary = true
         with patch.object(Watchdog, 'activate', Mock(return_value=False)):
             self.assertEqual(self.ha.run_cycle(), 'Demoting self because watchdog could not be activated')
             self.p.is_primary = false
@@ -616,6 +617,7 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
         self.e.initialize = true
         self.ha.bootstrap()
+        self.p.is_primary = true
         with patch.object(Watchdog, 'activate', Mock(return_value=False)), \
                 patch('patroni.ha.logger.error') as mock_logger:
             self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap')
@@ -689,110 +691,289 @@ class TestHa(PostgresInit):
 
     @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
     def test_manual_failover_from_leader(self):
+        self.ha.has_lock = true  # I am the leader
+
+        # to me
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', self.p.name, None))
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            mock_warning.assert_called_with('%s: I am already the leader, no need to %s', 'manual failover', 'failover')
+
+        # to a non-existent candidate
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla', None))
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            mock_warning.assert_called_with(
+                '%s: no healthy members found, %s is not possible', 'manual failover', 'failover')
+
+        # to an existent candidate
         self.ha.fetch_node_status = get_node_status()
-        self.ha.has_lock = true
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
-        f = Failover(0, self.p.name, '', None)
-        self.ha.cluster = get_cluster_initialized_with_leader(f)
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'b', None))
+        self.ha.cluster.members.append(Member(0, 'b', 28, {'api_url': 'http://127.0.0.1:8011/patroni'}))
         self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
+
+        # to a candidate on an older timeline
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.fetch_node_status = get_node_status(timeline=1)
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            self.assertEqual(mock_info.call_args_list[0][0],
+                             ('Timeline %s of member %s is behind the cluster timeline %s', 1, 'b', 2))
+
+        # to a lagging candidate
+        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})
+            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', 'b'))
+            self.ha.cluster.members.pop()
+
+    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    def test_manual_switchover_from_leader(self):
+        self.ha.has_lock = true  # I am the leader
+
+        self.ha.fetch_node_status = get_node_status()
+
+        # different leader specified in failover key, no candidate
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None))
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            mock_warning.assert_called_with(
+                '%s: leader name does not match: %s != %s', 'switchover', 'blabla', 'postgresql0')
+
+        # no candidate
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
+        self.assertEqual(self.ha.run_cycle(), 'switchover: demoting myself')
+
         self.ha._rewind.rewind_or_reinitialize_needed_and_possible = true
-        self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
-        self.ha.fetch_node_status = get_node_status(nofailover=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
-        self.ha.fetch_node_status = get_node_status(watchdog_failed=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
-        self.ha.fetch_node_status = get_node_status(timeline=1)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
-        self.ha.fetch_node_status = get_node_status(wal_position=1)
-        self.ha.cluster.config.data.update({'maximum_lag_on_failover': 5})
-        self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
-        # manual failover from the previous leader to us won't happen if we hold the nofailover flag
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'switchover: demoting myself')
 
-        # Failover scheduled time must include timezone
-        scheduled = datetime.datetime.now()
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.ha.run_cycle()
+        # other members with failover_limitation_s
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.fetch_node_status = get_node_status(nofailover=True)
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            self.assertEqual(mock_info.call_args_list[0][0], ('Member %s is %s', 'leader', 'not allowed to promote'))
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.fetch_node_status = get_node_status(watchdog_failed=True)
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            self.assertEqual(mock_info.call_args_list[0][0], ('Member %s is %s', 'leader', 'not watchdog capable'))
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.fetch_node_status = get_node_status(timeline=1)
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            self.assertEqual(mock_info.call_args_list[0][0],
+                             ('Timeline %s of member %s is behind the cluster timeline %s', 1, 'leader', 2))
+        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})
+            self.ha.global_config = self.ha.patroni.config.get_global_config(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'))
 
+    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    def test_scheduled_switchover_from_leader(self):
+        self.ha.has_lock = true  # I am the leader
+
+        self.ha.fetch_node_status = get_node_status()
+
+        # switchover scheduled time must include timezone
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            scheduled = datetime.datetime.now()
+            self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'blabla', scheduled))
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
+            self.assertIn('Incorrect value of scheduled_at: %s', mock_warning.call_args_list[0][0])
+
+        # scheduled now
         scheduled = datetime.datetime.utcnow().replace(tzinfo=tzutc)
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'b', scheduled))
+        self.ha.cluster.members.append(Member(0, 'b', 28, {'api_url': 'http://127.0.0.1:8011/patroni'}))
+        self.assertEqual('switchover: demoting myself', self.ha.run_cycle())
 
-        scheduled = scheduled + datetime.timedelta(seconds=30)
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+        # scheduled in the future
+        with patch('patroni.ha.logger.info') as mock_info:
+            scheduled = scheduled + datetime.timedelta(seconds=30)
+            self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'blabla', scheduled))
+            self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+            self.assertIn('Awaiting %s at %s (in %.0f seconds)', mock_info.call_args_list[0][0])
 
-        scheduled = scheduled + datetime.timedelta(seconds=-600)
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+        # stale value
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            scheduled = scheduled + datetime.timedelta(seconds=-600)
+            self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'b', scheduled))
+            self.ha.cluster.members.append(Member(0, 'b', 28, {'api_url': 'http://127.0.0.1:8011/patroni'}))
+            self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+            self.assertIn('Found a stale %s value, cleaning up: %s', mock_warning.call_args_list[0][0])
 
-        scheduled = None
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+    def test_manual_switchover_from_leader_in_pause(self):
+        self.ha.has_lock = true  # I am the leader
+        self.ha.is_paused = true
+
+        # no candidate
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+            mock_warning.assert_called_with(
+                '%s is possible only to a specific candidate in a paused state', 'Switchover')
 
     def test_manual_failover_from_leader_in_pause(self):
         self.ha.has_lock = true
+        self.ha.fetch_node_status = get_node_status()
         self.ha.is_paused = true
-        scheduled = datetime.datetime.now()
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
-        self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+
+        # failover from me, candidate is healthy
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, None, 'b', None))
+        self.ha.cluster.members.append(Member(0, 'b', 28, {'api_url': 'http://127.0.0.1:8011/patroni'}))
+        self.assertEqual('PAUSE: manual failover: demoting myself', self.ha.run_cycle())
+        self.ha.cluster.members.pop()
 
     def test_manual_failover_from_leader_in_synchronous_mode(self):
-        self.ha.has_lock = true
         self.ha.is_synchronous_mode = true
         self.ha.process_sync_replication = Mock()
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, None))
-        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, 'a'))
-        self.ha.is_failover_possible = true
+        self.ha.fetch_node_status = get_node_status()
+
+        # I am the leader
+        self.p.is_primary = true
+        self.ha.has_lock = true
+
+        # the candidate is not in sync members but we allow failover to an async candidate
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, None, 'b', None), sync=(self.p.name, 'a'))
+        self.ha.cluster.members.append(Member(0, 'b', 28, {'api_url': 'http://127.0.0.1:8011/patroni'}))
         self.assertEqual('manual failover: demoting myself', self.ha.run_cycle())
+        self.ha.cluster.members.pop()
+
+    def test_manual_switchover_from_leader_in_synchronous_mode(self):
+        self.ha.is_synchronous_mode = true
+        self.ha.process_sync_replication = Mock()
+
+        # I am the leader
+        self.p.is_primary = true
+        self.ha.has_lock = true
+
+        # candidate specified is not in sync members
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None),
+                                                                  sync=(self.p.name, 'blabla'))
+            self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+            self.assertEqual(mock_warning.call_args_list[0][0],
+                             ('%s candidate=%s does not match with sync_standbys=%s', 'Switchover', 'a', 'blabla'))
+
+        # the candidate is in sync members and is healthy
+        self.ha.fetch_node_status = get_node_status(wal_position=305419896)
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None),
+                                                              sync=(self.p.name, 'a'))
+        self.ha.cluster.members.append(Member(0, 'a', 28, {'api_url': 'http://127.0.0.1:8011/patroni'}))
+        self.assertEqual('switchover: demoting myself', self.ha.run_cycle())
+
+        # the candidate is in sync members but is not healthy
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.fetch_node_status = get_node_status(nofailover=true)
+            self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
+            self.assertEqual(mock_info.call_args_list[0][0], ('Member %s is %s', 'a', 'not allowed to promote'))
 
     def test_manual_failover_process_no_leader(self):
         self.p.is_primary = false
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', self.p.name, None))
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None))
         self.p.set_role('replica')
-        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-        self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
-        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, self.p.name, '', None))
-        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
-        self.ha.fetch_node_status = get_node_status(reachable=False)  # inaccessible, in_recovery
+
+        # failover to another member, fetch_node_status for candidate fails
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None))
+            self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+            self.assertEqual(mock_warning.call_args_list[1][0],
+                             ('%s: member %s is %s', 'manual failover', 'leader', 'not reachable'))
+
+        # failover to another member, candidate is accessible, in_recovery
         self.p.set_role('replica')
-        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-        # set failover flag to True for all members of the cluster
+        self.ha.fetch_node_status = get_node_status()
+        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
+
+        # set nofailover flag to True for all members of the cluster
         # this should elect the current member, as we are not going to call the API for it.
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None))
-        self.ha.fetch_node_status = get_node_status(nofailover=True)  # accessible, in_recovery
-        self.p.set_role('replica')
+        self.ha.fetch_node_status = get_node_status(nofailover=True)
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-        # same as previous, but set the current member to nofailover. In no case it should be elected as a leader
+
+        # failover to me but I am set to nofailover. In no case I should be elected as a leader
+        self.p.set_role('replica')
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None))
         self.ha.patroni.nofailover = True
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote')
 
+        self.ha.patroni.nofailover = False
+
+        # failover to another member that is on an older timeline (only failover_limitation() is checked)
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'b', None))
+            self.ha.cluster.members.append(Member(0, 'b', 28, {'api_url': 'http://127.0.0.1:8011/patroni'}))
+            self.ha.fetch_node_status = get_node_status(timeline=1)
+            self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
+            mock_info.assert_called_with('%s: to %s, i am %s', 'manual failover', 'b', 'postgresql0')
+
+        # failover to another member lagging behind the cluster_lsn (only failover_limitation() is checked)
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.cluster.config.data.update({'maximum_lag_on_failover': 5})
+            self.ha.fetch_node_status = get_node_status(wal_position=1)
+            self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
+            mock_info.assert_called_with('%s: to %s, i am %s', 'manual failover', 'b', 'postgresql0')
+
+    def test_manual_switchover_process_no_leader(self):
+        self.p.is_primary = false
+        self.p.set_role('replica')
+
+        # I was the leader, other members are healthy
+        self.ha.fetch_node_status = get_node_status()
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, self.p.name, '', None))
+        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
+
+        # I was the leader, I am the only healthy member
+        with patch('patroni.ha.logger.info') as mock_info:
+            self.ha.fetch_node_status = get_node_status(reachable=False)  # inaccessible, in_recovery
+            self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+            self.assertEqual(mock_info.call_args_list[0][0], ('Member %s is %s', 'leader', 'not reachable'))
+            self.assertEqual(mock_info.call_args_list[1][0], ('Member %s is %s', 'other', 'not reachable'))
+
     def test_manual_failover_process_no_leader_in_synchronous_mode(self):
         self.ha.is_synchronous_mode = true
         self.p.is_primary = false
+        self.ha.fetch_node_status = get_node_status(nofailover=True)  # other nodes are not healthy
 
-        # switchover to a specific node, which name doesn't match our name (postgresql0)
+        # manual failover when our name (postgresql0) isn't in the /sync key and the candidate node is not available
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None),
+                                                                 sync=('leader1', 'blabla'))
+        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
+
+        # manual failover when the candidate node isn't available but our name is in the /sync key
+        # while other sync node is nofailover
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None),
+                                                                     sync=('leader1', 'postgresql0'))
+            self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet()))
+            self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
+            self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+            self.assertEqual(mock_warning.call_args_list[0][0],
+                             ('%s: member %s is %s', 'manual failover', 'other', 'not allowed to promote'))
+
+        # manual failover to our node (postgresql0),
+        # which name is not in sync nodes list (some sync nodes are available)
+        self.p.set_role('replica')
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None),
+                                                                 sync=('leader1', 'other'))
+        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['leader1']),
+                                                               CaseInsensitiveSet(['leader1'])))
+        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+
+    def test_manual_switchover_process_no_leader_in_synchronous_mode(self):
+        self.ha.is_synchronous_mode = true
+        self.p.is_primary = false
+
+        # to a specific node, which name doesn't match our name (postgresql0)
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'other', None))
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
 
-        # switchover to our node (postgresql0), which name is not in sync nodes list
+        # to our node (postgresql0), which name is not in sync nodes list
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'postgresql0', None),
                                                                  sync=('leader1', 'blabla'))
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
 
-        # switchover from a specific leader, but our name (postgresql0) is not in the sync nodes list
+        # without candidate, our name (postgresql0) is not in the sync nodes list
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', '', None),
                                                                  sync=('leader', 'blabla'))
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
@@ -802,45 +983,31 @@ class TestHa(PostgresInit):
                                                                  sync=('postgresql0'))
         self.ha.patroni.nofailover = True
         self.assertEqual(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote')
-        self.ha.patroni.nofailover = False
-
-        # manual failover when our name (postgresql0) isn't in the /sync key and the `other` node is not available
-        self.ha.fetch_node_status = get_node_status(nofailover=True)  # accessible, in_recovery
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None),
-                                                                 sync=('leader1', 'blabla'))
-        self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
-
-        # manual failover when the `other` node isn't available but our name is in the /sync key
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None),
-                                                                 sync=('leader1', 'postgresql0'))
-        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet()))
-        self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
-        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-
-        # manual failover to our node (postgresql0),
-        # which name is not in sync nodes list (the leader and all sync nodes are not available)
-        self.p.set_role('replica')
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None),
-                                                                 sync=('leader1', 'other'))
-        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-
-        # manual failover to our node (postgresql0),
-        # which name is not in sync nodes list (some sync nodes are available)
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None),
-                                                                 sync=('leader1', 'other'))
-        self.p.set_role('replica')
-        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['leader1']),
-                                                               CaseInsensitiveSet(['leader1'])))
-        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
 
     def test_manual_failover_process_no_leader_in_pause(self):
         self.ha.is_paused = true
+
+        # I am running as primary, cluster is unlocked, the candidate is allowed to promote
+        # but we are in pause
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None))
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: continue to run as primary without lock')
+
+    def test_manual_switchover_process_no_leader_in_pause(self):
+        self.ha.is_paused = true
+
+        # I am running as primary, cluster is unlocked, no candidate specified
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', '', None))
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: continue to run as primary without lock')
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'blabla', None))
-        self.assertEqual('PAUSE: acquired session lock as a leader', self.ha.run_cycle())
+
+        # the candidate is not running
+        with patch('patroni.ha.logger.warning') as mock_warning:
+            self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'blabla', None))
+            self.assertEqual('PAUSE: acquired session lock as a leader', self.ha.run_cycle())
+            self.assertEqual(
+                mock_warning.call_args_list[0][0],
+                ('%s: removing failover key because failover candidate is not running', 'switchover'))
+
+        # switchover to me, I am not leader
         self.p.is_primary = false
         self.p.set_role('replica')
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', self.p.name, None))
@@ -848,7 +1015,7 @@ class TestHa(PostgresInit):
 
     def test_is_healthiest_node(self):
         self.ha.is_failsafe_mode = true
-        self.p.is_primary = false
+        self.ha.state_handler.is_primary = false
         self.ha.patroni.nofailover = False
         self.ha.fetch_node_status = get_node_status()
         self.ha.dcs._last_failsafe = {'foo': ''}
@@ -1090,7 +1257,7 @@ class TestHa(PostgresInit):
         f = Failover(0, self.p.name, '', None)
         self.ha.cluster = get_cluster_initialized_with_leader(f)
         self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
-        self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
+        self.assertEqual(self.ha.run_cycle(), 'switchover: demoting myself')
 
     @patch('patroni.ha.Ha.demote')
     def test_failover_immediately_on_zero_primary_start_timeout(self, demote):