Fixes and improvements in failsafe (#2532)

1. Fix problem with logical slots not advancing when only the primary lost access to DCS
2. Don't let Patroni to join as a raft voting member when running failsafe behave tests. It allows to test exactly the same conditions as for other DCS
3. Speed up dcs_failsafe_mode behave tests by getting rid from long sleeps, slight reshuffling of places when we start/stop outage, and by killing Patroni/Postgres to avoid long shutdown due to the leader key removal attempts.

features/dcs_failsafe_mode.feature

@@ -24,9 +24,9 @@ Feature: dcs failsafe mode
 
   @dcs-failsafe
   Scenario: check new replica isn't promoted when leader is down and DCS is up
+    Given 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
@@ -47,31 +47,28 @@ Feature: dcs failsafe mode
   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 Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 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
+    And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 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 kill postgres1
+    And I kill postmaster on 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
+    Given I shut down postgres0
+    And DCS is up
     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
+    And postgres1 role is the primary after 25 seconds
 
   @dcs-failsafe
   Scenario: check three-node cluster is functioning while DCS is down
@@ -80,8 +77,9 @@ Feature: dcs failsafe mode
     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
+    And replication works from postgres1 to postgres0 after 10 seconds
     Given DCS is down
-    And I sleep for 12 seconds
+    Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 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

features/environment.py

@@ -184,7 +184,9 @@ class PatroniController(AbstractController):
             config.pop('etcd', None)
 
         raft_port = os.environ.get('RAFT_PORT')
-        if raft_port:
+        # If patroni_raft_controller is suspended two Patroni members is enough to get a quorum,
+        # therefore we don't want Patroni to join as a voting member when testing dcs_failsafe_mode.
+        if raft_port and not self._output_dir.endswith('dcs_failsafe_mode'):
             os.environ['RAFT_PORT'] = str(int(raft_port) + 1)
             config['raft'] = {'data_dir': self._output_dir, 'self_addr': 'localhost:' + os.environ['RAFT_PORT']}
 

patroni/ha.py

@@ -105,7 +105,6 @@ class Failsafe(object):
             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
@@ -1641,31 +1640,27 @@ class Ha(object):
                 # try to start dead postgres
                 return self.recover()
 
-            try:
-                if self.cluster.is_unlocked():
-                    ret = self.process_unhealthy_cluster()
-                else:
-                    msg = self.process_healthy_cluster()
-                    ret = self.evaluate_scheduled_restart() or msg
-            finally:
-                # we might not have a valid PostgreSQL connection here if another thread
-                # stops PostgreSQL, therefore, we only reload replication slots if no
-                # asynchronous processes are running (should be always the case for the master)
-                if not self._async_executor.busy and not self.state_handler.is_starting():
-                    create_slots = self.state_handler.slots_handler.sync_replication_slots(self.cluster,
-                                                                                           self.patroni.nofailover,
-                                                                                           self.patroni.replicatefrom,
-                                                                                           self.is_paused())
-                    if not self.state_handler.cb_called:
-                        if not self.state_handler.is_leader():
-                            self._rewind.trigger_check_diverged_lsn()
-                        self.state_handler.call_nowait(ACTION_ON_START)
-                    if create_slots and self.cluster.leader:
-                        err = self._async_executor.try_run_async('copy_logical_slots',
-                                                                 self.state_handler.slots_handler.copy_logical_slots,
-                                                                 args=(self.cluster, create_slots))
-                        if not err:
-                            ret = 'Copying logical slots {0} from the primary'.format(create_slots)
+            if self.cluster.is_unlocked():
+                ret = self.process_unhealthy_cluster()
+            else:
+                msg = self.process_healthy_cluster()
+                ret = self.evaluate_scheduled_restart() or msg
+
+            # we might not have a valid PostgreSQL connection here if another thread
+            # stops PostgreSQL, therefore, we only reload replication slots if no
+            # asynchronous processes are running (should be always the case for the master)
+            if not self._async_executor.busy and not self.state_handler.is_starting():
+                create_slots = self._sync_replication_slots(False)
+                if not self.state_handler.cb_called:
+                    if not self.state_handler.is_leader():
+                        self._rewind.trigger_check_diverged_lsn()
+                    self.state_handler.call_nowait(ACTION_ON_START)
+                if create_slots and self.cluster.leader:
+                    err = self._async_executor.try_run_async('copy_logical_slots',
+                                                             self.state_handler.slots_handler.copy_logical_slots,
+                                                             args=(self.cluster, create_slots))
+                    if not err:
+                        ret = 'Copying logical slots {0} from the primary'.format(create_slots)
             return ret
         except DCSError:
             dcs_failed = True
@@ -1675,9 +1670,9 @@ class Ha(object):
             return 'Error communicating with PostgreSQL. Will try again later'
         finally:
             if not dcs_failed:
-                self.touch_member()
-                if self.state_handler.is_leader():
+                if self.is_leader():
                     self._failsafe.set_is_active(0)
+                self.touch_member()
 
     def _handle_dcs_error(self):
         if not self.is_paused() and self.state_handler.is_running():
@@ -1694,15 +1689,35 @@ class Ha(object):
                 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'
-            elif self.is_failsafe_mode():
-                cluster = self._failsafe.update_cluster(self.cluster)
-                if cluster:
-                    self.state_handler.slots_handler.sync_replication_slots(cluster,
+            else:
+                self._sync_replication_slots(True)
+        return 'DCS is not accessible'
+
+    def _sync_replication_slots(self, dcs_failed):
+        """Handles replication slots.
+
+        :param dcs_failed: bool, indicates that communication with DCS failed (get_cluster() or update_leader())
+        :returns: list[str], replication slots names that should be copied from the primary"""
+
+        slots = []
+
+        # If dcs_failed we don't want to touch replication slots on a leader or replicas if failsafe_mode isn't enabled.
+        if not self.cluster or dcs_failed and (self.is_leader() or not self.is_failsafe_mode()):
+            return slots
+
+        # It could be that DCS is read-only, or only the leader can't access it.
+        # Only the second one could be handled by `load_cluster_from_dcs()`.
+        # The first one affects advancing logical replication slots on replicas, therefore we rely on
+        # Failsafe.update_cluster(), that will return "modified" Cluster if failsafe mode is active.
+        cluster = self._failsafe.update_cluster(self.cluster)\
+            if self.is_failsafe_mode() and not self.is_leader() else self.cluster
+        if cluster:
+            slots = self.state_handler.slots_handler.sync_replication_slots(cluster,
                                                                             self.patroni.nofailover,
                                                                             self.patroni.replicatefrom,
                                                                             self.is_paused())
-
-        return 'DCS is not accessible'
+        # Don't copy replication slots if failsafe_mode is active
+        return [] if self.failsafe_is_active() else slots
 
     def run_cycle(self):
         with self._async_executor:

tests/test_ha.py

@@ -500,6 +500,14 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(),
                          'continue to run as a leader because failsafe mode is enabled and all members are accessible')
 
+    def test_readonly_dcs_primary_failsafe(self):
+        self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe()
+        self.ha.dcs.update_leader = Mock(side_effect=DCSError('Etcd is not responding properly'))
+        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()
@@ -508,6 +516,14 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible')
 
+    def test_no_dcs_connection_replica_failsafe_not_enabled_but_active(self):
+        self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        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')