Repository: zalando/patroni
Pull request #2379 (2022-08-10 13:13:15)

DCS failsafe mode
If enabled it will allow Patroni to cope with DCS outages.
In case of a DCS outage the leader tries to call all remaining members in the cluster via API and if all of them respond with success the leader will not be demoted.

The failsafe_mode could be enabled by running
```sh
patronictl edit-config -s failsafe_mode=true
```

or by calling the `/config` REST API endpoint.

CyberDem0n: @thedodd yes and no.
This implementation is trying to make a difference between the reasons why the update of the leader lock failed and the failsafe mechanism triggers only when we got a network communication exception or some sort of InternalError exception from DCS.
Right now it is hard to guarantee that all possible corner cases are covered.

The #2318 simply considers any failed attempt to update the leader lock as a failsafe trigger and should not bring many surprises. Also, it is a very common request to have a Patroni just to handle configuration management (yeah, I know, it defeats the purpose of Patroni :) )
thedodd: @CyberDem0n just some quick feedback. I've created a docker image to test these changes, and my setup is as follows:
- All code is present as of this branch's latest commit `772fe19`.
- I've configured a 2 member cluster, with failsafe_mode=true, running in a K8s cluster.
- When I query their /failsafe endpoint, they return:
  ```
    "member0": "http://10.22.27.41:8008/patroni",
    "member1": "http://10.22.10.123:8008/patroni"
  ```
- Cluster is stable and has a master & a replica.

If I then delete the K8s RoleBinding which grants Patroni access to the K8s API (for updating endpoints/services/labels &c), it looks like the primary is still demoting.

Technically, the K8s API will have returned a 403 (which I see in the Patroni logs). Does this circumvent the failsafe logic? Upon reviewing the code, it is not immediately obvious to me.
CyberDem0n: > If I then delete the K8s RoleBinding which grants Patroni access to the K8s API (for updating endpoints/services/labels &c), it looks like the primary is still demoting.
Technically, the K8s API will have returned a 403 (which I see in the Patroni logs). Does this circumvent the failsafe logic? Upon reviewing the code, it is not immediately obvious to me.

Hi @thedodd. The failsafe logic is triggered when DCS doesn't respond in time or when it is asking to repeat the request after some timeout. But, when retrying there could still be the same issue.
In both cases Patroni is giving up after `retry_timeout` seconds and the failsafe mode is triggered. The 403 status code only indicates that DCS rejects requests from the current node, but in fact, it is healthy. I can't immediately tell whether it would be safe to use 403 in order to trigger the failsafe mechanism.
CyberDem0n: @alexeyklyukin yeah, nodes must be able to communicate with each other using credentials and/or client certificates.
Sure, we can call the `POST /failsafe` in order to make sure that authentication succeeds, but there are still many questions:
1. Should we completely ignore the failsafe mode in this case or just don't add the node to the list?
2. On the next iteration the check will be executed again. It will not only spam logs, but cause some increased resource usage.
3. What if the check succeeds initially but starts failing later? For example, someone changed credentials on the other node (it is not easy to detect). Or maybe some certificates expired.
thedodd: @CyberDem0n ok, quick update: by way of manually crippling the K8s API server pods of the testing cluster, I was able to induce the needed downtime scenario. Failsafe mode appears to have held the Patroni cluster steady (1 master, 1 replica). I could connect, reconnect, and continue to use the master as expected. No cluster topology changes during the control plane downtime.
anyafit: hi, i try use this PR and get error in communicating with DCS, but i can't see that this feature is work
CyberDem0n: > What is my mistake?

@anyafit for existing clusters one should use `patronictl edit-config` to enable it: https://github.com/zalando/patroni/blob/e54f736651743a2521bdf393f067ea191ae08e19/docs/dcs_failsafe_mode.rst
You can also check whether the feature is enabled using `patronictl show-config`.
CyberDem0n: > by way of manually crippling the K8s API server pods of the testing cluster, I was able to induce the needed downtime scenario. Failsafe mode appears to have held the Patroni cluster steady

@thedodd about simple cases I am quite confident that it works. We have behave tests for that, that send SIGSTOP to the localkube process in order to simulate outage.
thedodd: @CyberDem0n thoughts on when you would like to land this PR? Waiting on a few other reviews?
CyberDem0n: @thedodd it should be more sophisticated than just stopping the control plane.
For example:
- slow network
- overloaded control-plane
- leader not being able to access the control plane, with replicas not having this problem
OlleLarsson: Hi @CyberDem0n, any updates on this feature 🤗 ?
CyberDem0n: @OlleLarsson more people will test it - faster it will be merged and released. Feel free to participate :)
OlleLarsson: > @OlleLarsson more people will test it - faster it will be merged and released. Feel free to participate :)

Perhaps I can find some time.
Is there are specific scenario that's extra interesting to test?
CyberDem0n: @OlleLarsson ideally it should be something close to real problems with DCS.
anyafit: hi, i have some problem with connect to etcd. It's mean that problem solved too fast?```
2022-10-27 20:18:44,385 INFO: Lock owner: patroni_hostname; I am patroni_hostname
2022-10-27 20:18:47,724 ERROR: Request to server http://etcd_hostname:2379 failed: ReadTimeoutError("HTTPConnectionPool(host='etcd_hostname', port=2379): Read timed out. (read timeout=3.333218111967047)",)
2022-10-27 20:18:47,724 INFO: Reconnection allowed, looking for another server.
2022-10-27 20:18:47,724 INFO: Retrying on http://etcd_hostname_2:2379
2022-10-27 20:18:47,907 INFO: Selected new etcd server http://etcd_hostname_2:2379
2022-10-27 20:18:47,908 ERROR: watchprefix failed: ProtocolError('Connection broken: IncompleteRead(0 bytes read)', IncompleteRead(0 bytes read))
2022-10-27 20:18:48,401 INFO: no action. I am (patroni_hostname), the leader with the lock
2022-10-27 20:18:49,577 ERROR: Request to server http://etcd_hostname_2:2379 failed: MaxRetryError("HTTPConnectionPool(host='etcd_hostname_2', port=2379): Max retries exceeded with url: /v3/kv/range (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x7f48d2f91c50>, 'Connection to etcd_hostname_2 timed out. (connect timeout=1.6666666666666667)'))",)
2022-10-27 20:18:49,577 INFO: Reconnection allowed, looking for another server.
2022-10-27 20:18:49,577 INFO: Retrying on http://etcd_hostname:2379
2022-10-27 20:18:49,755 INFO: Selected new etcd server http://etcd_hostname:2379
2022-10-27 20:18:55,924 INFO: no action. I am (patroni_hostname), the leader with the lock

```
hughcapet: 👍 
CyberDem0n: :+1:

Patch:
diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst
index 3c783654..37d40add 100644
--- a/docs/SETTINGS.rst
+++ b/docs/SETTINGS.rst
@@ -21,6 +21,7 @@ Dynamic configuration is stored in the DCS (Distributed Configuration Store) and
 - **master\_stop\_timeout**: The number of seconds Patroni is allowed to wait when stopping Postgres and effective only when synchronous_mode is enabled. When set to > 0 and the synchronous_mode is enabled, Patroni sends SIGKILL to the postmaster if the stop operation is running for more than the value set by master_stop_timeout. Set the value according to your durability/availability tradeoff. If the parameter is not set or set <= 0, master_stop_timeout does not apply.
 -  **synchronous\_mode**: turns on synchronous replication mode. In this mode a replica will be chosen as synchronous and only the latest leader and synchronous replica are able to participate in leader election. Synchronous mode makes sure that successfully committed transactions will not be lost at failover, at the cost of losing availability for writes when Patroni cannot ensure transaction durability. See :ref:`replication modes documentation <replication_modes>` for details.
 -  **synchronous\_mode\_strict**: prevents disabling synchronous replication if no synchronous replicas are available, blocking all client writes to the primary. See :ref:`replication modes documentation <replication_modes>` for details.
+-  **failsafe\_mode**: Enables :ref:`DCS Failsafe Mode <dcs_failsafe_mode>`. Defaults to `false`.
 -  **postgresql**:
     -  **use\_pg\_rewind**: whether or not to use pg_rewind. Defaults to `false`.
     -  **use\_slots**: whether or not to use replication slots. Defaults to `true` on PostgreSQL 9.4+.
diff --git a/docs/dcs_failsafe_mode.rst b/docs/dcs_failsafe_mode.rst
new file mode 100644
index 00000000..4f7b669e
--- /dev/null
+++ b/docs/dcs_failsafe_mode.rst
@@ -0,0 +1,63 @@
+.. _dcs_failsafe_mode:
+
+DCS Failsafe Mode
+=================
+
+The problem
+-----------
+
+Patroni is heavily relying on Distributed Configuration Store (DCS) to solve the task of leader elections and detect network partitioning. That is, the node is allowed to run Postgres as the primary only if it can update the leader lock in DCS. In case the update of the leader lock fails, Postgres is immediately demoted and started as read-only. Depending on which DCS is used, the chances of hitting the "problem" differ. For example, with Etcd which is only used for Patroni, chances are close to zero, while with K8s API (backed by Etcd) it could be observed more frequently.
+
+
+Reasons for the current implementation
+---------------------------------------
+
+The leader lock update failure could be caused by two main reasons:
+
+1. Network partitioning
+2. DCS being down
+
+In general, it is impossible to distinguish between these two from a single node, and therefore Patroni assumes the worst case - network partitioning. In the case of a partitioned network, other nodes of the Patroni cluster may successfully grab the leader lock and promote Postgres to primary. In order to avoid a split-brain, the old primary is demoted before the leader lock expires.
+
+
+DCS Failsafe Mode
+-----------------
+
+We introduce a new special option, the ``failsafe_mode``. It could be enabled only via global configuration stored in the DCS ``/config`` key. If the failsafe mode is enabled and the leader lock update in DCS failed due to reasons different from the version/value/index mismatch, Postgres may continue to run as a primary if it can access all known members of the cluster via Patroni REST API.
+
+
+Low-level implementation details
+--------------------------------
+
+- We introduce a new, permanent key in DCS, named ``/failsafe``.
+- The ``/failsafe`` key contains all known members of the given Patroni cluster at a given time.
+- The current leader maintains the ``/failsafe`` key.
+- The member is allowed to participate in the leader race and become the new leader only if it is present in the ``/failsafe`` key.
+- If the cluster consists of a single node the ``/failsafe`` key will contain a single member.
+- In the case of DCS "outage" the existing primary connects to all members presented in the ``/failsafe`` key via the ``POST /failsafe`` REST API and may continue to run as the primary if all replicas acknowledge it.
+- If one of the members doesn't respond, the primary is demoted.
+- Replicas are using incoming ``POST /failsafe`` REST API requests as an indicator that the primary is still alive. This information is cached for ``ttl`` seconds.
+
+
+F.A.Q.
+------
+
+- Why MUST the current primary see ALL other members? Can’t we rely on quorum here?
+
+  This is a great question! The problem is that the view on the quorum might be different from the perspective of DCS and Patroni. While DCS nodes must be evenly distributed across availability zones, there is no such rule for Patroni, and more importantly, there is no mechanism for introducing and enforcing such a rule. If the majority of Patroni nodes ends up in the losing part of the partitioned network (including primary) while minority nodes are in the winning part, the primary must be demoted. Only checking ALL other members allows detecting such a situation.
+
+- What if node/pod gets terminated while DCS is down?
+
+  If DCS isn’t accessible, the check “are ALL other cluster members accessible?” is executed every cycle of the heartbeat loop (every ``loop_wait`` seconds). If pod/node is terminated, the check will fail and Postgres will be demoted to a read-only and will not recover until DCS is restored.
+
+- What if all members of the Patroni cluster are lost while DCS is down?
+
+  Patroni could be configured to create the new replica from the backup even when the cluster doesn't have a leader. But, if the new member isn't present in the ``/failsafe`` key, it will not be able to grab the leader lock and promote.
+  
+- What will happen if the primary lost access to DCS while replicas didn't?
+
+  The primary will execute the failsafe code and contact all known replicas. These replicas will use this information as an indicator that the primary is alive and will not start the leader race even if the leader lock in DCS has expired.
+
+- How to enable the Failsafe Mode?
+
+  Before enabling the ``failsafe_mode`` please make sure that Patroni version on all members is up-to-date. After that, you can use either the ``PATCH /config`` :ref:`REST API <rest_api>` or ``patronictl edit-config -s failsafe_mode=true``
diff --git a/docs/index.rst b/docs/index.rst
index 6753e53a..008ea54a 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -21,6 +21,7 @@ Currently supported PostgreSQL versions: 9.3 to 15.
 
    README
    dynamic_configuration
+   dcs_failsafe_mode
    rest_api
    existing_data
    ENVIRONMENT
diff --git a/features/dcs_failsafe_mode.feature b/features/dcs_failsafe_mode.feature
new file mode 100644
index 00000000..52278041
--- /dev/null
+++ b/features/dcs_failsafe_mode.feature
@@ -0,0 +1,87 @@
+Feature: dcs failsafe mode
+  We should check the basic dcs failsafe mode functioning
+
+  Scenario: check failsafe mode can be successfully enabled
+    Given I start postgres0
+    And postgres0 is a leader after 10 seconds
+    And I sleep for 3 seconds
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"loop_wait": 2, "ttl": 20, "retry_timeout": 5, "failsafe_mode": true}
+    Then I receive a response code 200
+    And Response on GET http://127.0.0.1:8008/failsafe contains postgres0 after 10 seconds
+    When I issue a GET request to http://127.0.0.1:8008/failsafe
+    Then I receive a response code 200
+    And I receive a response postgres0 http://127.0.0.1:8008/patroni
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}}}
+    Then I receive a response code 200
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"dcs_slot_0": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
+    Then I receive a response code 200
+
+  @dcs-failsafe
+  Scenario: check one-node cluster is functioning while DCS is down
+    Given DCS is down
+    Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds
+    And postgres0 role is the primary after 10 seconds
+
+  @dcs-failsafe
+  Scenario: check new replica isn't promoted when leader is down and DCS is up
+    When I do a backup of postgres0
+    And I shut down postgres0
+    And DCS is up
+    When I start postgres1 in a cluster batman from backup with no_master
+    And I sleep for 2 seconds
+    Then postgres1 role is the replica after 12 seconds
+
+  Scenario: check leader and replica are both in /failsafe key after leader is back
+    Given I start postgres0
+    And I start postgres1
+    Then "members/postgres0" key in DCS has state=running after 10 seconds
+    And "members/postgres1" key in DCS has state=running after 2 seconds
+    And Response on GET http://127.0.0.1:8009/failsafe contains postgres1 after 10 seconds
+    When I issue a GET request to http://127.0.0.1:8009/failsafe
+    Then I receive a response code 200
+    And I receive a response postgres0 http://127.0.0.1:8008/patroni
+    And I receive a response postgres1 http://127.0.0.1:8009/patroni
+
+  @dcs-failsafe
+  @slot-advance
+  Scenario: check leader and replica are functioning while DCS is down
+    Given logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds
+    And DCS is down
+    And I sleep for 12 seconds
+    Then postgres0 role is the primary after 10 seconds
+    And postgres1 role is the replica after 2 seconds
+    And replication works from postgres0 to postgres1 after 10 seconds
+    And I get all changes from logical slot dcs_slot_0 on postgres0
+    And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds
+
+  @dcs-failsafe
+  Scenario: check master is demoted when one replica is shut down and DCS is down
+    Given DCS is down
+    And I shut down postgres1
+    And I sleep for 2 seconds
+    Then postgres0 role is the replica after 12 seconds
+
+  @dcs-failsafe
+  Scenario: check known replica is promoted when leader is down and DCS is up
+    Given DCS is up
+    Then postgres0 role is the primary after 22 seconds
+    When I start postgres1
+    Then "members/postgres1" key in DCS has state=running after 10 seconds
+    And Response on GET http://127.0.0.1:8009/failsafe contains postgres1 after 10 seconds
+    Given DCS is down
+    And I shut down postgres0
+    And DCS is up
+    Then postgres1 role is the primary after 22 seconds
+
+  @dcs-failsafe
+  Scenario: check three-node cluster is functioning while DCS is down
+    Given I start postgres0
+    And I start postgres2
+    Then "members/postgres0" key in DCS has state=running after 10 seconds
+    And "members/postgres2" key in DCS has state=running after 10 seconds
+    And Response on GET http://127.0.0.1:8008/failsafe contains postgres2 after 10 seconds
+    Given DCS is down
+    And I sleep for 12 seconds
+    Then postgres1 role is the primary after 10 seconds
+    And postgres0 role is the replica after 2 seconds
+    And postgres2 role is the replica after 2 seconds
diff --git a/features/environment.py b/features/environment.py
index 6a45de29..b1a76f5a 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -2,6 +2,7 @@ import abc
 import datetime
 import os
 import json
+import psutil
 import re
 import shutil
 import signal
@@ -239,7 +240,22 @@ class PatroniController(AbstractController):
             self.recursive_update(config, custom_config)
 
         self.recursive_update(config, {
-            'bootstrap': {'dcs': {'loop_wait': 2, 'postgresql': {'parameters': {'wal_keep_segments': 100}}}}})
+            'bootstrap': {
+                'dcs': {
+                    'loop_wait': 2,
+                    'postgresql': {
+                        'parameters': {
+                            'wal_keep_segments': 100,
+                            'archive_mode': 'on',
+                            'archive_command': (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT +
+                                                ' --mode archive ' +
+                                                '--dirname {} --filename %f --pathname %p').format(
+                                                    os.path.join(self._work_directory, 'data', 'wal_archive'))
+                        }
+                    }
+                }
+            }
+        })
         if config['postgresql'].get('callbacks', {}).get('on_role_change'):
             config['postgresql']['callbacks']['on_role_change'] += ' ' + str(self.__PORT)
 
@@ -355,6 +371,7 @@ class AbstractDcsController(AbstractController):
 
     def __init__(self, context, mktemp=True):
         work_directory = mktemp and tempfile.mkdtemp() or None
+        self._paused = False
         super(AbstractDcsController, self).__init__(context, self.name(), work_directory, context.pctl.output_dir)
 
     def _is_accessible(self):
@@ -366,6 +383,16 @@ class AbstractDcsController(AbstractController):
         if self._work_directory:
             shutil.rmtree(self._work_directory)
 
+    def start_outage(self):
+        if not self._paused and self._handle:
+            self._handle.suspend()
+            self._paused = True
+
+    def stop_outage(self):
+        if self._paused and self._handle:
+            self._handle.resume()
+            self._paused = False
+
     def path(self, key=None, scope='batman'):
         return self._CLUSTER_NODE.format(scope) + (key and '/' + key or '')
 
@@ -404,8 +431,8 @@ class ConsulController(AbstractDcsController):
         self._config_file = self._work_directory + '.json'
         with open(self._config_file, 'wb') as f:
             f.write(b'{"session_ttl_min":"5s","server":true,"bootstrap":true,"advertise_addr":"127.0.0.1"}')
-        return subprocess.Popen(['consul', 'agent', '-config-file', self._config_file, '-data-dir',
-                                 self._work_directory], stdout=self._log, stderr=subprocess.STDOUT)
+        return psutil.Popen(['consul', 'agent', '-config-file', self._config_file, '-data-dir',
+                             self._work_directory], stdout=self._log, stderr=subprocess.STDOUT)
 
     def stop(self, kill=False, timeout=15):
         super(ConsulController, self).stop(kill=kill, timeout=timeout)
@@ -441,8 +468,8 @@ class AbstractEtcdController(AbstractDcsController):
         self._client_cls = client_cls
 
     def _start(self):
-        return subprocess.Popen(["etcd", "--enable-v2=true", "--data-dir", self._work_directory],
-                                stdout=self._log, stderr=subprocess.STDOUT)
+        return psutil.Popen(["etcd", "--enable-v2=true", "--data-dir", self._work_directory],
+                            stdout=self._log, stderr=subprocess.STDOUT)
 
     def _is_running(self):
         from patroni.dcs.etcd import DnsCachingResolver
@@ -499,7 +526,43 @@ class Etcd3Controller(AbstractEtcdController):
             assert False, "exception when cleaning up etcd contents: {0}".format(e)
 
 
-class KubernetesController(AbstractDcsController):
+class AbstractExternalDcsController(AbstractDcsController):
+
+    def __init__(self, context, mktemp=True):
+        super(AbstractExternalDcsController, self).__init__(context, mktemp)
+        self._wrapper = ['sudo']
+
+    def _start(self):
+        return self._external_pid
+
+    def start_outage(self):
+        if not self._paused:
+            subprocess.call(self._wrapper + ['kill', '-SIGSTOP', self._external_pid])
+            self._paused = True
+
+    def stop_outage(self):
+        if self._paused:
+            subprocess.call(self._wrapper + ['kill', '-SIGCONT', self._external_pid])
+            self._paused = False
+
+    def _has_started(self):
+        return True
+
+    @abc.abstractmethod
+    def process_name():
+        """process name to search with pgrep"""
+
+    def _is_running(self):
+        if not self._handle:
+            self._external_pid = subprocess.check_output(['pgrep', '-nf', self.process_name()]).decode('utf-8').strip()
+            return False
+        return True
+
+    def stop(self):
+        pass
+
+
+class KubernetesController(AbstractExternalDcsController):
 
     def __init__(self, context):
         super(KubernetesController, self).__init__(context)
@@ -515,10 +578,37 @@ class KubernetesController(AbstractDcsController):
         self._client = k8s_client
         self._api = self._client.CoreV1Api()
 
-    def _start(self):
-        pass
+    def process_name(self):
+        return "localkube"
+
+    def _is_running(self):
+        if not self._handle:
+            context = os.environ.get('PATRONI_KUBERNETES_CONTEXT')
+            if context.startswith('kind-'):
+                container = '{0}-control-plane'.format(context[5:])
+                api_process = 'kube-apiserver'
+            elif context.startswith('k3d-'):
+                container = '{0}-server-0'.format(context)
+                api_process = 'k3s'
+            else:
+                return super(KubernetesController, self)._is_running()
+            try:
+                docker = 'docker'
+                with open(os.devnull, 'w') as null:
+                    if subprocess.call([docker, 'info'], stdout=null, stderr=null) != 0:
+                        raise Exception
+            except Exception:
+                docker = 'podman'
+                with open(os.devnull, 'w') as null:
+                    if subprocess.call([docker, 'info'], stdout=null, stderr=null) != 0:
+                        raise Exception
+            self._wrapper = [docker, 'exec', container]
+            self._external_pid = subprocess.check_output(self._wrapper + ['pidof', api_process]).decode('utf-8').strip()
+            return False
+        return True
 
     def create_pod(self, name, scope):
+        self.delete_pod(name)
         labels = self._labels.copy()
         labels['cluster-name'] = scope
         metadata = self._client.V1ObjectMeta(namespace=self._namespace, name=name, labels=labels)
@@ -567,11 +657,8 @@ class KubernetesController(AbstractDcsController):
             if len(result.items) < 1:
                 break
 
-    def _is_running(self):
-        return True
 
-
-class ZooKeeperController(AbstractDcsController):
+class ZooKeeperController(AbstractExternalDcsController):
 
     """ handles all zookeeper related tasks, used for the tests setup and cleanup """
 
@@ -583,8 +670,8 @@ class ZooKeeperController(AbstractDcsController):
         import kazoo.client
         self._client = kazoo.client.KazooClient()
 
-    def _start(self):
-        pass  # TODO: implement later
+    def process_name(self):
+        return "zookeeper"
 
     def query(self, key, scope='batman'):
         import kazoo.exceptions
@@ -603,6 +690,9 @@ class ZooKeeperController(AbstractDcsController):
             assert False, "exception when cleaning up zookeeper contents: {0}".format(e)
 
     def _is_running(self):
+        if not super(ZooKeeperController, self)._is_running():
+            return False
+
         # if zookeeper is running, but we didn't start it
         if self._client.connected:
             return True
@@ -652,9 +742,9 @@ class RaftController(AbstractDcsController):
         del env['PATRONI_RAFT_PARTNER_ADDRS']
         env['PATRONI_RAFT_SELF_ADDR'] = self.CONTROLLER_ADDR
         env['PATRONI_RAFT_DATA_DIR'] = self._work_directory
-        return subprocess.Popen([sys.executable, '-m', 'coverage', 'run',
-                                '--source=patroni', '-p', 'patroni_raft_controller.py'],
-                                stdout=self._log, stderr=subprocess.STDOUT, env=env)
+        return psutil.Popen([sys.executable, '-m', 'coverage', 'run',
+                             '--source=patroni', '-p', 'patroni_raft_controller.py'],
+                            stdout=self._log, stderr=subprocess.STDOUT, env=env)
 
     def query(self, key, scope='batman'):
         ret = self._raft.get(self.path(key, scope))
@@ -683,6 +773,7 @@ class PatroniPoolController(object):
 
     PYTHON = sys.executable.replace('\\', '/')
     BACKUP_SCRIPT = [PYTHON, 'features/backup_create.py']
+    BACKUP_RESTORE_SCRIPT = ' '.join((PYTHON, os.path.abspath('features/backup_restore.py'))).replace('\\', '/')
     ARCHIVE_RESTORE_SCRIPT = ' '.join((PYTHON, os.path.abspath('features/archive-restore.py')))
 
     def __init__(self, context):
@@ -769,7 +860,7 @@ class PatroniPoolController(object):
                     'archive_mode': 'on',
                     'archive_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode archive ' +
                                         '--dirname {} --filename %f --pathname %p').format(
-                                        os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/'))
+                                        os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/'))
                 },
                 'authentication': {
                     'superuser': {'password': 'zalando1'},
@@ -785,14 +876,14 @@ class PatroniPoolController(object):
             'bootstrap': {
                 'method': 'backup_restore',
                 'backup_restore': {
-                    'command': (self.PYTHON + ' features/backup_restore.py --sourcedir=' +
+                    'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' +
                                 os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')),
                     'recovery_conf': {
                         'recovery_target_action': 'promote',
                         'recovery_target_timeline': 'latest',
                         'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' +
                                             '--dirname {} --filename %f --pathname %p').format(
-                                            os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/'))
+                            os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/'))
                     }
                 }
             },
@@ -805,6 +896,25 @@ class PatroniPoolController(object):
         }
         self.start(name, custom_config=custom_config)
 
+    def bootstrap_from_backup_no_master(self, name, cluster_name):
+        custom_config = {
+            'scope': cluster_name,
+            'postgresql': {
+                'recovery_conf': {
+                    'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' +
+                                        '--dirname {} --filename %f --pathname %p').format(
+                                        os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/'))
+                },
+                'create_replica_methods': ['no_master_bootstrap'],
+                'no_master_bootstrap': {
+                    'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' +
+                                os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')),
+                    'no_master': '1'
+                }
+            }
+        }
+        self.start(name, custom_config=custom_config)
+
     @property
     def dcs(self):
         if self._dcs is None:
@@ -981,7 +1091,9 @@ def before_feature(context, feature):
 
 
 def after_feature(context, feature):
-    """ stop all Patronis, remove their data directory and cleanup the keys in etcd """
+    """ send SIGCONT to a dcs if neccessary,
+    stop all Patronis remove their data directory and cleanup the keys in etcd """
+    context.dcs_ctl.stop_outage()
     context.pctl.stop_all()
     data = os.path.join(context.pctl.patroni_path, 'data')
     if os.path.exists(data):
@@ -997,3 +1109,5 @@ def before_scenario(context, scenario):
             if p._conn and p._conn.server_version < 110000:
                 scenario.skip('pg_replication_slot_advance() is not supported on {0}'.format(p._conn.server_version))
                 break
+    if 'dcs-failsafe' in scenario.effective_tags and not context.dcs_ctl._handle:
+        scenario.skip('it is not possible to control state of {0} from tests'.format(context.dcs_ctl.name()))
diff --git a/features/steps/dcs_failsafe_mode.py b/features/steps/dcs_failsafe_mode.py
new file mode 100644
index 00000000..613d698f
--- /dev/null
+++ b/features/steps/dcs_failsafe_mode.py
@@ -0,0 +1,16 @@
+from behave import step
+
+
+@step('DCS is down')
+def start_dcs_outage(context):
+    context.dcs_ctl.start_outage()
+
+
+@step('DCS is up')
+def stop_dcs_outage(context):
+    context.dcs_ctl.stop_outage()
+
+
+@step('I start {name:w} in a cluster {cluster_name:w} from backup with no_master')
+def start_cluster_from_backup_no_master(context, name, cluster_name):
+    context.pctl.bootstrap_from_backup_no_master(name, cluster_name)
diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py
index 54cd72e1..a745b192 100644
--- a/features/steps/patroni_api.py
+++ b/features/steps/patroni_api.py
@@ -109,6 +109,8 @@ def check_response(context, component, data):
         assert data.strip('"') in context.response, "response {0} does not contain {1}".format(context.response, data)
     else:
         assert component in context.response, "{0} is not part of the response".format(component)
+        if context.certfile:
+            data = data.replace('http://', 'https://')
         assert str(context.response[component]) == str(data), "{0} does not contain {1}".format(component, data)
 
 
diff --git a/patroni/api.py b/patroni/api.py
index ff44b515..1988b1ba 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -302,6 +302,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
         metrics.append("# TYPE patroni_cluster_unlocked gauge")
         metrics.append("patroni_cluster_unlocked{0} {1}".format(scope_label, int(postgres.get('cluster_unlocked', 0))))
 
+        metrics.append("# HELP patroni_failsafe_mode_is_active Value is 1 if the cluster is unlocked, 0 if locked.")
+        metrics.append("# TYPE patroni_failsafe_mode_is_active gauge")
+        metrics.append("patroni_failsafe_mode_is_active{0} {1}"
+                       .format(scope_label, int(postgres.get('failsafe_mode_is_active', 0))))
+
         metrics.append("# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_timeline counter")
         metrics.append("patroni_postgres_timeline{0} {1}".format(scope_label, postgres.get('timeline', 0)))
@@ -368,6 +373,24 @@ class RestApiHandler(BaseHTTPRequestHandler):
         self.server.patroni.sighup_handler()
         self._write_response(202, 'reload scheduled')
 
+    def do_GET_failsafe(self):
+        failsafe = self.server.patroni.dcs.failsafe
+        if isinstance(failsafe, dict):
+            self._write_json_response(200, failsafe)
+        else:
+            self.send_error(502)
+
+    @check_access
+    def do_POST_failsafe(self):
+        if self.server.patroni.ha.is_failsafe_mode():
+            request = self._read_json_content()
+            if request:
+                message = self.server.patroni.ha.update_failsafe(request) or 'Accepted'
+                code = 200 if message == 'Accepted' else 500
+                self._write_response(code, message)
+        else:
+            self.send_error(502)
+
     @check_access
     def do_POST_sigterm(self):
         """Only for behave testing on windows"""
@@ -670,6 +693,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         if not cluster or cluster.is_unlocked():
             result['cluster_unlocked'] = True
+        if self.server.patroni.ha.failsafe_is_active():
+            result['failsafe_mode_is_active'] = True
         result['dcs_last_seen'] = self.server.patroni.dcs.last_seen
         return result
 
diff --git a/patroni/config.py b/patroni/config.py
index 9e1019f7..d0f01eb4 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -68,6 +68,7 @@ class Config(object):
         'synchronous_mode': False,
         'synchronous_mode_strict': False,
         'synchronous_node_count': 1,
+        'failsafe_mode': False,
         'standby_cluster': {
             'create_replica_methods': '',
             'host': '',
@@ -235,7 +236,7 @@ class Config(object):
                     if name in self.__DEFAULT_CONFIG['standby_cluster']:
                         config['standby_cluster'][name] = deepcopy(value)
             elif name in config:  # only variables present in __DEFAULT_CONFIG allowed to be overridden from DCS
-                if name in ('synchronous_mode', 'synchronous_mode_strict'):
+                if name in ('synchronous_mode', 'synchronous_mode_strict', 'failsafe_mode'):
                     config[name] = value
                 else:
                     config[name] = int(value)
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 10242f0d..c27ffa50 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -458,6 +458,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,f
     :param sync: reference to `SyncState` object, last observed synchronous replication state.
     :param history: reference to `TimelineHistory` object
     :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}
+    :param failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list.
     """
 
     @property
@@ -830,6 +831,10 @@ class AbstractDCS(object):
                 and self._write_failsafe(json.dumps(value, separators=(',', ':'))):
             self._last_failsafe = value
 
+    @property
+    def failsafe(self):
+        return self._last_failsafe
+
     @abc.abstractmethod
     def _update_leader(self):
         """Update leader key (or session) ttl
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index 67541a34..c3bbd0bc 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -852,7 +852,8 @@ class Kubernetes(AbstractDCS):
             except (TypeError, ValueError):
                 ttl = self._ttl
 
-            if not metadata or not self._leader_observed_time or self._leader_observed_time + ttl < time.time():
+            if not metadata or not self._leader_observed_time or self._leader_observed_time + ttl < time.time() \
+                    and (self._name != leader or not isinstance(failsafe, dict) or leader not in failsafe):
                 leader = None
 
             if metadata:
diff --git a/patroni/ha.py b/patroni/ha.py
index 53801107..62836a41 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -18,7 +18,7 @@ from .postgresql import ACTION_ON_START, ACTION_ON_ROLE_CHANGE
 from .postgresql.misc import postgres_version_to_int
 from .postgresql.rewind import Rewind
 from .utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster, parse_int
-from .dcs import RemoteMember
+from .dcs import Cluster, Leader, RemoteMember
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +70,65 @@ class _MemberStatus(namedtuple('_MemberStatus', ['member', 'reachable', 'in_reco
         return None
 
 
+class Failsafe(object):
+
+    def __init__(self, dcs):
+        self._lock = RLock()
+        self._dcs = dcs
+        self._last_update = 0
+        self._name = None
+        self._conn_url = None
+        self._api_url = None
+        self._slots = None
+
+    def update(self, data):
+        with self._lock:
+            self._last_update = time.time()
+            self._name = data['name']
+            self._conn_url = data['conn_url']
+            self._api_url = data['api_url']
+            self._slots = data.get('slots')
+
+    @property
+    def leader(self):
+        with self._lock:
+            if self._last_update + self._dcs.ttl > time.time():
+                return Leader(None, None,
+                              RemoteMember(self._name, {'api_url': self._api_url,
+                                                        'conn_url': self._conn_url,
+                                                        'slots': self._slots}))
+
+    def update_cluster(self, cluster):
+        # Enreach cluster with the real leader if there was a ping from it
+        leader = self.leader
+        if leader:
+            cluster = list(cluster)
+            # We rely on the strict order of fields in the namedtuple
+            cluster[2] = leader
+            cluster[4].append(leader.member)
+            cluster[8] = leader.member.data['slots']
+            cluster = Cluster(*cluster)
+        return cluster
+
+    def is_active(self):
+        """Is used to report in REST API whether the failsafe mode was activated.
+
+           On primary the self._last_update is set from the
+           set_is_active() method and always returns the correct value.
+
+           On replicas the self._last_update is set at the moment when
+           the primary performs POST /failsafe REST API calls.
+           The side-effect - it is possible that replicas will show
+           failsafe_is_active values different from the primary."""
+
+        with self._lock:
+            return self._last_update + self._dcs.ttl > time.time()
+
+    def set_is_active(self, value):
+        with self._lock:
+            self._last_update = value
+
+
 class Ha(object):
 
     def __init__(self, patroni):
@@ -81,6 +140,7 @@ class Ha(object):
         self.old_cluster = None
         self._is_leader = False
         self._is_leader_lock = RLock()
+        self._failsafe = Failsafe(patroni.dcs)
         self._was_paused = False
         self._leader_timeline = None
         self.recovering = False
@@ -149,6 +209,10 @@ class Ha(object):
             self.old_cluster = cluster
         self.cluster = cluster
 
+        if self.cluster.is_unlocked() and self.is_failsafe_mode():
+            # If failsafe mode is enabled we want to inject the "real" leader to the cluster
+            self.cluster = cluster = self._failsafe.update_cluster(cluster)
+
         if not self.has_lock(False):
             self.set_is_leader(False)
 
@@ -165,6 +229,13 @@ class Ha(object):
         self.set_is_leader(ret)
         return ret
 
+    def _failsafe_config(self):
+        if self.is_failsafe_mode():
+            ret = {m.name: m.api_url for m in self.cluster.members}
+            if self.state_handler.name not in ret:
+                ret[self.state_handler.name] = self.patroni.api.connection_string
+            return ret
+
     def update_lock(self, write_leader_optime=False):
         last_lsn = slots = None
         if write_leader_optime:
@@ -174,7 +245,7 @@ class Ha(object):
             except Exception:
                 logger.exception('Exception when called state_handler.last_operation()')
         try:
-            ret = self.dcs.update_leader(last_lsn, slots)
+            ret = self.dcs.update_leader(last_lsn, slots, self._failsafe_config())
         except DCSError:
             raise
         except Exception:
@@ -483,6 +554,9 @@ class Ha(object):
     def is_synchronous_mode_strict(self):
         return self.check_mode('synchronous_mode_strict')
 
+    def is_failsafe_mode(self):
+        return self.check_mode('failsafe_mode')
+
     def process_sync_replication(self):
         """Process synchronous standby beahvior.
 
@@ -681,6 +755,49 @@ class Ha(object):
         pool.join()
         return results
 
+    def update_failsafe(self, data):
+        if self.state_handler.state == 'running' and self.state_handler.role == 'master':
+            return 'Running as a leader'
+        self._failsafe.update(data)
+
+    def failsafe_is_active(self):
+        return self._failsafe.is_active()
+
+    def call_failsafe_member(self, data, member):
+        try:
+            response = self.patroni.request(member, 'post', 'failsafe', data, timeout=2, retries=1)
+            data = response.data.decode('utf-8')
+            logger.info('Got response from %s %s: %s', member.name, member.api_url, data)
+            return response.status == 200 and data == 'Accepted'
+        except Exception as e:
+            logger.warning("Request failed to %s: POST %s (%s)", member.name, member.api_url, e)
+        return False
+
+    def check_failsafe_topology(self):
+        failsafe = self.dcs.failsafe
+        if not isinstance(failsafe, dict) or self.state_handler.name not in failsafe:
+            return False
+        data = {
+            'name': self.state_handler.name,
+            'conn_url': self.state_handler.connection_string,
+            'api_url': self.patroni.api.connection_string,
+        }
+        try:
+            data['slots'] = self.state_handler.slots()
+        except Exception:
+            logger.exception('Exception when called state_handler.slots()')
+        members = [RemoteMember(name, {'api_url': url})
+                   for name, url in failsafe.items()
+                   if name != self.state_handler.name]
+        if not members:  # A sinlge node cluster
+            return True
+        pool = ThreadPool(len(members))
+        call_failsafe_member = functools.partial(self.call_failsafe_member, data)
+        results = pool.map(call_failsafe_member, members)
+        pool.close()
+        pool.join()
+        return all(results)
+
     def is_lagging(self, wal_position):
         """Returns if instance with an wal should consider itself unhealthy to be promoted due to replication lag.
 
@@ -836,8 +953,19 @@ class Ha(object):
             logger.warning('Watchdog device is not usable')
             return False
 
+        all_known_members = self.old_cluster.members
+        if self.is_failsafe_mode():
+            failsafe_members = self.dcs.failsafe
+            # We want to discard failsafe_mode if the /failsafe key contains garbage or empty.
+            if isinstance(failsafe_members, dict):
+                # If current node is missing in the /failsafe key we immediately disqualify it from the race.
+                if failsafe_members and self.state_handler.name not in failsafe_members:
+                    return False
+                # Race among not only existing cluster members, but also all known members from the failsafe config
+                all_known_members += [RemoteMember(name, {'api_url': url}) for name, url in failsafe_members.items()]
+        all_known_members += self.cluster.members
+
         # When in sync mode, only last known master and sync standby are allowed to promote automatically.
-        all_known_members = self.cluster.members + self.old_cluster.members
         if self.is_synchronous_mode() and self.cluster.sync and self.cluster.sync.leader:
             if not self.cluster.sync.matches(self.state_handler.name):
                 return False
@@ -1542,19 +1670,39 @@ class Ha(object):
         except DCSError:
             dcs_failed = True
             logger.error('Error communicating with DCS')
-            if not self.is_paused() and self.state_handler.is_running() and self.state_handler.is_leader():
+            return self._handle_dcs_error()
+        except (psycopg.Error, PostgresConnectionException):
+            return 'Error communicating with PostgreSQL. Will try again later'
+        finally:
+            if not dcs_failed:
+                self.touch_member()
+                if self.state_handler.is_leader():
+                    self._failsafe.set_is_active(0)
+
+    def _handle_dcs_error(self):
+        if not self.is_paused() and self.state_handler.is_running():
+            if self.state_handler.is_leader():
+                if self.is_failsafe_mode() and self.check_failsafe_topology():
+                    self.set_is_leader(True)
+                    self._failsafe.set_is_active(time.time())
+                    self.watchdog.keepalive()
+                    return 'continue to run as a leader because failsafe mode is enabled and all members are accessible'
+                self._failsafe.set_is_active(0)
                 msg = 'demoting self because DCS is not accessible and I was a leader'
                 if not self._async_executor.try_run_async(msg, self.demote, ('offline',)):
                     return msg
                 logger.warning('AsyncExecutor is busy, demoting from the main thread')
                 self.demote('offline')
                 return 'demoted self because DCS is not accessible and I was a leader'
-            return 'DCS is not accessible'
-        except (psycopg.Error, PostgresConnectionException):
-            return 'Error communicating with PostgreSQL. Will try again later'
-        finally:
-            if not dcs_failed:
-                self.touch_member()
+            elif self.is_failsafe_mode():
+                cluster = self._failsafe.update_cluster(self.cluster)
+                if cluster:
+                    self.state_handler.slots_handler.sync_replication_slots(cluster,
+                                                                            self.patroni.nofailover,
+                                                                            self.patroni.replicatefrom,
+                                                                            self.is_paused())
+
+        return 'DCS is not accessible'
 
     def run_cycle(self):
         with self._async_executor:


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index a0fc8958..0da270d7 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -43,11 +43,13 @@ class MockResponse(object):
         return {'content-type': 'json'}
 
 
-def requests_get(url, **kwargs):
+def requests_get(url, method='GET', endpoint=None, data='', **kwargs):
     members = '[{"id":14855829450254237642,"peerURLs":["http://localhost:2380","http://localhost:7001"],' +\
               '"name":"default","clientURLs":["http://localhost:2379","http://localhost:4001"]}]'
     response = MockResponse()
-    if url.startswith('http://local'):
+    if endpoint == 'failsafe':
+        response.content = 'Accepted'
+    elif url.startswith('http://local'):
         raise urllib3.exceptions.HTTPError()
     elif ':8011/patroni' in url:
         response.content = '{"role": "replica", "wal": {"received_location": 0}, "tags": {}}'
@@ -56,7 +58,6 @@ def requests_get(url, **kwargs):
     elif url.startswith('http://exhibitor'):
         response.content = '{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}'
     elif url.endswith(':8011/reinitialize'):
-        data = kwargs.get('data', '')
         if ' false}' in data:
             response.status_code = 503
             response.content = 'restarting after failure already in progress'
diff --git a/tests/test_api.py b/tests/test_api.py
index 37228980..668d4156 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -61,6 +61,14 @@ class MockHa(object):
     state_handler = MockPostgresql()
     watchdog = MockWatchdog()
 
+    @staticmethod
+    def update_failsafe(*args):
+        return 'foo'
+
+    @staticmethod
+    def failsafe_is_active(*args):
+        return True
+
     @staticmethod
     def is_leader():
         return False
@@ -371,6 +379,20 @@ class TestRestApiHandler(unittest.TestCase):
         mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, config)
         MockRestApiServer(RestApiHandler, request)
 
+    @patch.object(MockPatroni, 'dcs')
+    def test_do_GET_failsafe(self, mock_dcs):
+        type(mock_dcs).failsafe = PropertyMock(return_value={'node1': 'http://foo:8080/patroni'})
+        self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /failsafe'))
+        type(mock_dcs).failsafe = PropertyMock(return_value=None)
+        self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /failsafe'))
+
+    def test_do_POST_failsafe(self):
+        with patch.object(MockHa, 'is_failsafe_mode', Mock(return_value=False), create=True):
+            self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization))
+        with patch.object(MockHa, 'is_failsafe_mode', Mock(return_value=True), create=True):
+            self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization +
+                                                                   '\nContent-Length: 9\n\n{"a":"b"}'))
+
     @patch.object(MockPatroni, 'sighup_handler', Mock())
     def test_do_POST_reload(self):
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization))
diff --git a/tests/test_etcd.py b/tests/test_etcd.py
index 61980474..0e70462a 100644
--- a/tests/test_etcd.py
+++ b/tests/test_etcd.py
@@ -40,7 +40,7 @@ def etcd_read(self, key, **kwargs):
         raise etcd.EtcdKeyNotFound
 
     response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [
-                {"key": "/service/batman5/config", "value": '{"synchronous_mode": 0}',
+                {"key": "/service/batman5/config", "value": '{"synchronous_mode": 0, "failsafe_mode": true}',
                  "modifiedIndex": 1582, "createdIndex": 1582},
                 {"key": "/service/batman5/failover", "value": "",
                  "modifiedIndex": 1582, "createdIndex": 1582},
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 92551982..c9c2f129 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -33,12 +33,12 @@ def false(*args, **kwargs):
     return False
 
 
-def get_cluster(initialize, leader, members, failover, sync, cluster_config=None):
+def get_cluster(initialize, leader, members, failover, sync, cluster_config=None, failsafe=None):
     t = datetime.datetime.now().isoformat()
     history = TimelineHistory(1, '[[1,67197376,"no recovery target specified","' + t + '","foo"]]',
                               [(1, 67197376, 'no recovery target specified', t, 'foo')])
     cluster_config = cluster_config or ClusterConfig(1, {'check_timeline': True}, 1)
-    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history, None, None)
+    return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history, None, failsafe)
 
 
 def get_cluster_not_initialized_without_leader(cluster_config=None):
@@ -49,7 +49,7 @@ def get_cluster_bootstrapping_without_leader(cluster_config=None):
     return get_cluster("", None, [], None, SyncState(None, None, None), cluster_config)
 
 
-def get_cluster_initialized_without_leader(leader=False, failover=None, sync=None, cluster_config=None):
+def get_cluster_initialized_without_leader(leader=False, failover=None, sync=None, cluster_config=None, failsafe=False):
     m1 = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres',
                                   'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4})
     leader = Leader(0, 0, m1 if leader else Member(0, '', 28, {}))
@@ -61,7 +61,8 @@ def get_cluster_initialized_without_leader(leader=False, failover=None, sync=Non
                                  'scheduled_restart': {'schedule': "2100-01-01 10:53:07.560445+00:00",
                                                        'postgres_version': '99.0.0'}})
     syncstate = SyncState(0 if sync else None, sync and sync[0], sync and sync[1])
-    return get_cluster(SYSID, leader, [m1, m2], failover, syncstate, cluster_config)
+    failsafe = {m.name: m.api_url for m in (m1, m2)} if failsafe else None
+    return get_cluster(SYSID, leader, [m1, m2], failover, syncstate, cluster_config, failsafe)
 
 
 def get_cluster_initialized_with_leader(failover=None, sync=None):
@@ -84,6 +85,11 @@ def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None):
     )
 
 
+def get_cluster_initialized_with_leader_and_failsafe():
+    return get_cluster_initialized_without_leader(leader=True, failsafe=True,
+                                                  cluster_config=ClusterConfig(1, {'failsafe_mode': True}, 1))
+
+
 def get_node_status(reachable=True, in_recovery=True, dcs_last_seen=0,
                     timeline=2, wal_position=10, nofailover=False,
                     watchdog_failed=False):
@@ -143,7 +149,7 @@ zookeeper:
         self.scheduled_restart = {'schedule': future_restart_time,
                                   'postmaster_start_time': str(postmaster_start_time)}
         self.watchdog = Watchdog(self.config)
-        self.request = lambda member, **kwargs: requests_get(member.api_url, **kwargs)
+        self.request = lambda *args, **kwargs: requests_get(args[0].api_url, *args[1:], **kwargs)
 
 
 def run_async(self, func, args=()):
@@ -205,6 +211,7 @@ class TestHa(PostgresInit):
             self.ha.load_cluster_from_dcs = Mock()
 
     def test_update_lock(self):
+        self.ha.is_failsafe_mode = true
         self.p.last_operation = Mock(side_effect=PostgresConnectionException(''))
         self.ha.dcs.update_leader = Mock(side_effect=[DCSError(''), Exception])
         self.assertRaises(DCSError, self.ha.update_lock)
@@ -458,12 +465,52 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader')
 
-    def test_no_etcd_connection_master_demote(self):
+    def test_no_dcs_connection_master_demote(self):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
         self.assertEqual(self.ha.run_cycle(), 'demoting self because DCS is not accessible and I was a leader')
         self.ha._async_executor.schedule('dummy')
         self.assertEqual(self.ha.run_cycle(), 'demoted self because DCS is not accessible and I was a leader')
 
+    def test_check_failsafe_topology(self):
+        self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
+        self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe()
+        self.ha.dcs._last_failsafe = self.ha.cluster.failsafe
+        self.assertEqual(self.ha.run_cycle(), 'demoting self because DCS is not accessible and I was a leader')
+        self.ha.state_handler.name = self.ha.cluster.leader.name
+        self.assertFalse(self.ha.failsafe_is_active())
+        self.assertEqual(self.ha.run_cycle(),
+                         'continue to run as a leader because failsafe mode is enabled and all members are accessible')
+        self.assertTrue(self.ha.failsafe_is_active())
+        with patch.object(Postgresql, 'slots', Mock(side_effect=Exception)):
+            self.ha.patroni.request = Mock(side_effect=Exception)
+            self.assertEqual(self.ha.run_cycle(), 'demoting self because DCS is not accessible and I was a leader')
+            self.assertFalse(self.ha.failsafe_is_active())
+        self.ha.dcs._last_failsafe.clear()
+        self.ha.dcs._last_failsafe[self.ha.cluster.leader.name] = self.ha.cluster.leader.member.api_url
+        self.assertEqual(self.ha.run_cycle(),
+                         'continue to run as a leader because failsafe mode is enabled and all members are accessible')
+
+    def test_no_dcs_connection_primary_failsafe(self):
+        self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
+        self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe()
+        self.ha.dcs._last_failsafe = self.ha.cluster.failsafe
+        self.ha.state_handler.name = self.ha.cluster.leader.name
+        self.assertEqual(self.ha.run_cycle(),
+                         'continue to run as a leader because failsafe mode is enabled and all members are accessible')
+
+    def test_no_dcs_connection_replica_failsafe(self):
+        self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
+        self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe()
+        self.ha.update_failsafe({'name': 'leader', 'api_url': 'http://127.0.0.1:8008/patroni',
+                                 'conn_url': 'postgres://127.0.0.1:5432/postgres', 'slots': {'foo': 1000}})
+        self.p.is_leader = false
+        self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible')
+
+    def test_update_failsafe(self):
+        self.assertRaises(Exception, self.ha.update_failsafe, {})
+        self.p.set_role('master')
+        self.assertEqual(self.ha.update_failsafe({}), 'Running as a leader')
+
     @patch('time.sleep', Mock())
     def test_bootstrap_from_another_member(self):
         self.ha.cluster = get_cluster_initialized_with_leader()
@@ -725,10 +772,15 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: promoted self to leader by acquiring session lock')
 
     def test_is_healthiest_node(self):
+        self.ha.is_failsafe_mode = true
         self.ha.state_handler.is_leader = false
         self.ha.patroni.nofailover = False
         self.ha.fetch_node_status = get_node_status()
+        self.ha.dcs._last_failsafe = {'foo': ''}
+        self.assertFalse(self.ha.is_healthiest_node())
+        self.ha.dcs._last_failsafe = {'postgresql0': ''}
         self.assertTrue(self.ha.is_healthiest_node())
+        self.ha.dcs._last_failsafe = None
         with patch.object(Watchdog, 'is_healthy', PropertyMock(return_value=False)):
             self.assertFalse(self.ha.is_healthiest_node())
         with patch('patroni.postgresql.Postgresql.is_starting', return_value=True):
diff --git a/tests/test_patroni.py b/tests/test_patroni.py
index ee79e0e7..2b2094a0 100644
--- a/tests/test_patroni.py
+++ b/tests/test_patroni.py
@@ -143,7 +143,8 @@ class TestPatroni(unittest.TestCase):
         self.p.api.start = Mock()
         self.p.logger.start = Mock()
         self.p.config._dynamic_configuration = {}
-        self.assertRaises(SleepException, self.p.run)
+        with patch('patroni.dcs.Cluster.is_unlocked', Mock(return_value=True)):
+            self.assertRaises(SleepException, self.p.run)
         with patch('patroni.config.Config.reload_local_configuration', Mock(return_value=False)):
             self.p.sighup_handler()
             self.assertRaises(SleepException, self.p.run)