Possibility to disable automatic failover cluster-wide

Any node of the cluster will maintain it's member key until Patroni is running there. Master node will also maintain the leader key until postgres is running as a master. If there is not postgres or it is running 'in_recovery', Patroni will release leader lock. Bootstrap of a new cluster will work (it is possible to specify paused: true) in the `bootstrap.dcs`. Replicas also will be able to join the cluster if the leader lock exist. If the postgres is not running on the node it will not try to bring it up. Also it disables reinitialize and all kind of scheduled actions, i.e. scheduled restart and scheduled failover. In case if DCS stops being reachable Patroni will not "demote" master if the automatic failover was disabled. Patroni will not stop postgres on exit.
2026-08-26 23:50:23 +00:00 · 2016-08-26 10:51:43 +02:00
parent 93b9046aed
commit ac49835a3c
2 changed files with 106 additions and 40 deletions
@@ -27,8 +27,7 @@ class Ha(object):
        self._async_executor = AsyncExecutor()

    def is_paused(self):
-        return self.cluster and self.cluster.config and 'pause' in self.cluster.config.data \
-               and self.cluster.config.data['pause']
+        return self.cluster and self.cluster.config and self.cluster.config.data.get('pause', False)

    def load_cluster_from_dcs(self):
        cluster = self.dcs.get_cluster()
@@ -142,10 +141,17 @@ class Ha(object):
        if recovery:
            ret = demote_reason if self.has_lock() else follow_reason
        else:
-            ret = demote_reason if self.state_handler.is_leader() else follow_reason
+            is_leader = self.state_handler.is_leader()
+            ret = demote_reason if is_leader else follow_reason

        node_to_follow = self._get_node_to_follow(self.cluster)

+        if self.is_paused():
+            if is_leader:
+                return 'continue to run as master without lock'
+            elif not node_to_follow:
+                return 'no action'
+
        self.state_handler.follow(node_to_follow, self.cluster.leader, recovery, self._async_executor)

        return ret
@@ -227,6 +233,8 @@ class Ha(object):
        if failover.candidate:  # manual failover to specific member
            if failover.candidate == self.state_handler.name:  # manual failover to me
                return True
+            elif self.is_paused():
+                return False

            # find specific node and check that it is healthy
            member = self.cluster.get_member(failover.candidate, fallback_to_leader=False)
@@ -243,6 +251,8 @@ class Ha(object):

            # at this point we should consider all members as a candidates for failover
            # 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
        if failover.leader:
@@ -261,9 +271,15 @@ class Ha(object):
        return self._is_healthiest_node(members, check_replication_lag=False)

    def is_healthiest_node(self):
+        if self.is_paused() and self.cluster.failover and not self.cluster.failover.scheduled_at:
+            return self.manual_failover_process_no_leader()
+
        if self.state_handler.is_leader():  # leader is always the healthiest
            return True

+        if self.is_paused():
+            return False
+
        if self.patroni.nofailover:  # nofailover tag makes node always unhealthy
            return False

@@ -323,28 +339,35 @@ class Ha(object):
    def process_manual_failover_from_leader(self):
        failover = self.cluster.failover

+        if failover.scheduled_at and self.is_paused():
+            return
+
        if (failover.scheduled_at and not
            self.should_run_scheduled_action("failover", failover.scheduled_at, lambda:
-                                             self.dcs.manual_failover('', '', index=self.cluster.failover.index))):
+                                             self.dcs.manual_failover('', '', index=failover.index))):
            return

        if not failover.leader or failover.leader == self.state_handler.name:
            if not failover.candidate or failover.candidate != self.state_handler.name:
-                members = [m for m in self.cluster.members if not failover.candidate or m.name == failover.candidate]
-                if self.is_failover_possible(members):  # check that there are healthy members
-                    self._async_executor.schedule('manual failover: demote')
-                    self._async_executor.run_async(self.demote)
-                    return 'manual failover: demoting myself'
+                if not failover.candidate and self.is_paused():
+                    logger.warning('Failover is possible only to a specific candidate in a paused state')
                else:
-                    logger.warning('manual failover: no healthy members found, failover is not possible')
+                    members = [m for m in self.cluster.members
+                               if not failover.candidate or m.name == failover.candidate]
+                    if self.is_failover_possible(members):  # check that there are healthy members
+                        self._async_executor.schedule('manual failover: demote')
+                        self._async_executor.run_async(self.demote)
+                        return 'manual failover: demoting myself'
+                    else:
+                        logger.warning('manual failover: no healthy members found, failover is not possible')
            else:
                logger.warning('manual failover: I am already the leader, no need to failover')
        else:
            logger.warning('manual failover: leader name does not match: %s != %s',
-                           self.cluster.failover.leader, self.state_handler.name)
+                           failover.leader, self.state_handler.name)

        logger.info('Trying to clean up failover key')
-        self.dcs.manual_failover('', '', index=self.cluster.failover.index)
+        self.dcs.manual_failover('', '', index=failover.index)

    def process_unhealthy_cluster(self):
        """Cluster has no leader key"""
@@ -363,7 +386,7 @@ class Ha(object):
            if self.patroni.nofailover:
                return self.follow('demoting self because I am not allowed to become master',
                                   'following a different leader because I am not allowed to promote')
-            return self.follow('demoting self because i am not the healthiest node',
+            return self.follow('demoting self because i am not the healthiest node',  # should not happen in real life
                               'following a different leader because i am not the healthiest node')

    def process_healthy_cluster(self):
@@ -373,6 +396,11 @@ class Ha(object):
                if msg is not None:
                    return msg

+            if self.is_paused() and not self.state_handler.is_leader():
+                self.dcs.delete_leader()
+                self.dcs.reset_cluster()
+                return 'removed leader lock because postgres is not running as master'
+
            if self.update_lock():
                return self.enforce_master_role('no action.  i am the leader with the lock',
                                                'promoted self to leader because i had the session lock')
@@ -386,6 +414,8 @@ class Ha(object):
                           'no action.  i am a secondary and i am following a leader', False)

    def evaluate_scheduled_restart(self):
+        if self.is_paused():
+            return None
        # restart if we need to
        restart_data = self.future_restart_scheduled()
        if restart_data:
@@ -496,7 +526,10 @@ class Ha(object):

    def process_scheduled_action(self):
        if self.reinitialize_scheduled():
-            if self.cluster.is_unlocked():
+            if self.is_paused():
+                logger.warning('Cluster is in a pause state, can not reinitialize')
+                self._async_executor.reset_scheduled_action()
+            elif self.cluster.is_unlocked():
                logger.error('Cluster has no leader, can not reinitialize')
                self._async_executor.reset_scheduled_action()
            elif self.has_lock():
@@ -533,26 +566,6 @@ class Ha(object):
            return 'failed to start postgres'
        return None

-    def pause_action(self):
-        if not self.state_handler.is_healthy():
-            return "Postgresql is not running. No action due to paused state"
-
-        if not self.state_handler.is_leader():
-            if self.has_lock():
-                self.dcs.delete_leader()
-            return "I'm secondary. No action due to paused state"
-
-        if self.has_lock():
-            if not self.update_lock():
-                # Either there is no connection to DCS or someone else acquired the lock
-                logger.error('failed to update leader lock')
-                self.load_cluster_from_dcs()
-            return "I'm the leader. Updating leader key"
-        elif self.cluster.is_unlocked():
-            if not self.acquire_lock():
-                return "Can't acquire the lock. No action due to paused state"
-            return "Cluster has no leader. Acquiring leader key"
-
    def _run_cycle(self):
        try:
            self.load_cluster_from_dcs()
@@ -569,9 +582,6 @@ class Ha(object):
            if self._async_executor.busy:
                return self.handle_long_action_in_progress()

-            if self.is_paused():
-                return self.pause_action()
-
            # we've got here, so any async action has finished. Check if we tried to recover and failed
            if self.recovering:
                self.recovering = False
@@ -588,7 +598,7 @@ class Ha(object):
            if self.state_handler.data_directory_empty():
                return self.bootstrap()  # new node
            # "bootstrap", but data directory is not empty
-            elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked():
+            elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked() and not self.is_paused():
                self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid)
            else:
                # check if we are allowed to join
@@ -599,6 +609,13 @@ class Ha(object):

            # try to start dead postgres
            if not self.state_handler.is_healthy():
+                if self.is_paused():
+                    if self.has_lock():
+                        self.dcs.delete_leader()
+                        self.dcs.reset_cluster()
+                        return 'removed leader lock because postgres is not running'
+                    else:
+                        return 'postgres is not running'
                msg = self.recover()
                if msg is not None:
                    return msg
@@ -621,12 +638,13 @@ class Ha(object):
                    self.state_handler.sync_replication_slots(self.cluster)
        except DCSError:
            logger.error('Error communicating with DCS')
-            if self.state_handler.is_running() and self.state_handler.is_leader():
+            if not self.is_paused() and self.state_handler.is_running() and self.state_handler.is_leader():
                self.demote(delete_leader=False)
                return 'demoted self because DCS is not accessible and i was a leader'
+            return 'DCS is not accessible'
        except (psycopg2.Error, PostgresConnectionException):
            logger.exception('Error communicating with PostgreSQL. Will try again later')

    def run_cycle(self):
        with self._async_executor:
-            return self._run_cycle()
+            return (self.is_paused() and 'PAUSE: ' or '') + self._run_cycle()
@@ -166,6 +166,9 @@ class TestHa(unittest.TestCase):
        self.ha.cluster = get_cluster_initialized_with_leader()
        self.assertEquals(self.ha.run_cycle(), 'starting as readonly because i had the session lock')

+    def test_do_not_recover_in_pause(self):
+        pass
+
    @patch('sys.exit', return_value=1)
    @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True))
    def test_sysid_no_match(self, exit_mock):
@@ -234,6 +237,13 @@ class TestHa(unittest.TestCase):
        self.ha.patroni.replicatefrom = "foo"
        self.assertEquals(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')

+    def test_follow_in_pause(self):
+        self.ha.cluster.is_unlocked = false
+        self.ha.is_paused = true
+        self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock')
+        self.p.is_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'PAUSE: no action')
+
    def test_no_etcd_connection_master_demote(self):
        self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
        self.assertEquals(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader')
@@ -272,6 +282,11 @@ class TestHa(unittest.TestCase):
        self.ha.run_cycle()
        self.assertIsNone(self.ha._async_executor.scheduled_action)

+        with patch.object(Ha, 'is_paused', true):
+            self.ha.schedule_reinitialize()
+            self.ha.run_cycle()
+            self.assertIsNone(self.ha._async_executor.scheduled_action)
+
        self.ha.cluster = get_cluster_initialized_with_leader()
        self.ha.has_lock = true
        self.ha.schedule_reinitialize()
@@ -345,6 +360,16 @@ class TestHa(unittest.TestCase):
        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
        self.assertEquals('no action.  i am the leader with the lock', self.ha.run_cycle())

+    @patch('requests.get', requests_get)
+    def test_manual_failover_from_leader_in_pause(self):
+        self.ha.has_lock = true
+        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.assertEquals('PAUSE: no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
+        self.assertEquals('PAUSE: no action.  i am the leader with the lock', self.ha.run_cycle())
+
    @patch('requests.get', requests_get)
    def test_manual_failover_process_no_leader(self):
        self.p.is_leader = false
@@ -370,11 +395,20 @@ class TestHa(unittest.TestCase):
        self.ha.patroni.nofailover = True
        self.assertEquals(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote')

+    def test_manual_failover_process_no_leader_in_pause(self):
+        self.ha.is_paused = true
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None))
+        self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock')
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', '', None))
+        self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock')
+
    def test_is_healthiest_node(self):
        self.ha.state_handler.is_leader = false
        self.ha.patroni.nofailover = False
        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {})
        self.assertTrue(self.ha.is_healthiest_node())
+        self.ha.is_paused = true
+        self.assertFalse(self.ha.is_healthiest_node())

    def test__is_healthiest_node(self):
        self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
@@ -451,3 +485,17 @@ class TestHa(unittest.TestCase):
        self.p._pending_restart = False
        self.assertFalse(self.ha.restart_matches("replica", "9.5.2", True))
        self.assertTrue(self.ha.restart_matches("replica", "9.5.2", False))
+
+    def test_process_healthy_cluster_in_pause(self):
+        self.p.is_leader = false
+        self.ha.is_paused = true
+        self.p.name = 'leader'
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.assertEquals(self.ha.run_cycle(), 'PAUSE: removed leader lock because postgres is not running as master')
+
+    def test_postgres_unhealthy_in_pause(self):
+        self.ha.is_paused = true
+        self.p.is_healthy = false
+        self.assertEquals(self.ha.run_cycle(), 'PAUSE: postgres is not running')
+        self.ha.has_lock = true
+        self.assertEquals(self.ha.run_cycle(), 'PAUSE: removed leader lock because postgres is not running')