From 333d292eb34e923c377b81081cf00a869b1772c8 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 5 Jul 2021 09:30:31 +0200 Subject: [PATCH] Handle DNS issues in Raft implementation (#1960) - Resolve Node IP for every connection attempt - Handle exception with connection failures due to failed resolve - Set PySyncObj DNS Cache timeouts aligned with `loop_wait` and `ttl` In addition to that, postpone the leader race for freshly started Raft nodes. It will help with the situation when the leader node was alone and demoted the Postgres and after that, the replica arrives, and quickly takes the leader lock without really performing the leader race. Close https://github.com/zalando/patroni/issues/1930, https://github.com/zalando/patroni/issues/1931 --- patroni/dcs/raft.py | 27 ++++++++++++++++++++++++--- patroni/ha.py | 25 +++++++++++++++++-------- tests/test_ha.py | 21 ++++++++++++++++++++- tests/test_raft.py | 15 ++++++++++++--- 4 files changed, 73 insertions(+), 15 deletions(-) diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index 76c65efb..90fcee2f 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -5,8 +5,10 @@ import threading import time from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON +from pysyncobj.dns_resolver import globalDnsResolver +from pysyncobj.node import TCPNode from pysyncobj.transport import TCPTransport, CONNECTION_STATE -from pysyncobj.utility import TcpUtility, UtilityException +from pysyncobj.utility import TcpUtility from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from ..utils import validate_directory @@ -20,6 +22,20 @@ class _TCPTransport(TCPTransport): super(_TCPTransport, self).__init__(syncObj, selfNode, otherNodes) self.setOnUtilityMessageCallback('members', syncObj.getMembers) + def _connectIfNecessarySingle(self, node): + try: + return super(_TCPTransport, self)._connectIfNecessarySingle(node) + except Exception as e: + logger.debug('Connection to %s failed: %r', node, e) + return False + + +def resolve_host(self): + return globalDnsResolver().resolve(self.host) + + +setattr(TCPNode, 'ip', property(resolve_host)) + class SyncObjUtility(object): @@ -30,7 +46,7 @@ class SyncObjUtility(object): def executeCommand(self, command): try: return self._utility.executeCommand(self.__node, command) - except UtilityException: + except Exception: return None def getMembers(self): @@ -99,7 +115,8 @@ class KVStoreTTL(DynMemberSyncObj): file_template = file_template.replace(':', '_') if os.name == 'nt' else file_template file_template = os.path.join(raft_data_dir, file_template) conf = SyncObjConf(password=config.get('password'), autoTick=False, appendEntriesUseBatch=False, - bindAddress=config.get('bind_addr'), commandsWaitLeader=config.get('commandsWaitLeader'), + bindAddress=config.get('bind_addr'), dnsFailCacheTime=(config.get('loop_wait') or 10), + dnsCacheTime=(config.get('ttl') or 30), commandsWaitLeader=config.get('commandsWaitLeader'), fullDumpFile=(file_template + '.dump' if self_addr else None), journalFile=(file_template + '.journal' if self_addr else None), onReady=on_ready, dynamicMembershipChange=True) @@ -281,6 +298,10 @@ class Raft(AbstractDCS): def set_retry_timeout(self, retry_timeout): self._sync_obj.set_retry_timeout(retry_timeout) + def reload_config(self, config): + super(Raft, self).reload_config(config) + globalDnsResolver().setTimeouts(self.ttl, self.loop_wait) + @staticmethod def member(key, value): return Member.from_node(value['index'], os.path.basename(key), None, value['value']) diff --git a/patroni/ha.py b/patroni/ha.py index ee1c78e8..42237d9b 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1219,9 +1219,6 @@ class Ha(object): self._delete_leader() return 'removed leader key after trying and failing to start postgres' return 'failed to start postgres' - self._crash_recovery_executed = False - if self._rewind.executed and not self._rewind.failed: - self._rewind.reset_state() return None def cancel_initialization(self): @@ -1345,12 +1342,24 @@ class Ha(object): if self.state_handler.bootstrapping: return self.post_bootstrap() - if self.recovering and not self._rewind.is_needed: + if self.recovering: self.recovering = False - # Check if we tried to recover and failed - msg = self.post_recover() - if msg is not None: - return msg + + if not self._rewind.is_needed: + # Check if we tried to recover from postgres crash and failed + msg = self.post_recover() + if msg is not None: + return msg + + # Reset some states after postgres successfully started up + self._crash_recovery_executed = False + if self._rewind.executed and not self._rewind.failed: + self._rewind.reset_state() + + # The Raft cluster without a quorum takes a bit of time to stabilize. + # Therefore we want to postpone the leader race if we just started up. + if self.cluster.is_unlocked() and self.dcs.__class__.__name__ == 'Raft': + return 'started as a secondary' # is data directory empty? if self.state_handler.data_directory_empty(): diff --git a/tests/test_ha.py b/tests/test_ha.py index 726addc3..ac520c9c 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -251,6 +251,15 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'starting as a secondary') self.assertEqual(self.ha.run_cycle(), 'failed to start postgres') + def test_recover_raft(self): + self.p.controldata = lambda: {'Database cluster state': 'in recovery', 'Database system identifier': SYSID} + self.p.is_running = false + self.p.follow = true + self.assertEqual(self.ha.run_cycle(), 'starting as a secondary') + self.p.is_running = true + self.ha.dcs.__class__.__name__ = 'Raft' + self.assertEqual(self.ha.run_cycle(), 'started as a secondary') + def test_recover_former_master(self): self.p.follow = false self.p.is_running = false @@ -279,7 +288,17 @@ class TestHa(PostgresInit): def test_recover_with_rewind(self): self.p.is_running = false self.ha.cluster = get_cluster_initialized_with_leader() - self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader') + self.ha.cluster.leader.member.data.update(version='2.0.2', role='master') + self.ha._rewind.pg_rewind = true + self.ha._rewind.check_leader_is_not_in_recovery = true + with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)): + self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader') + with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)): + self.p.follow = true + self.assertEqual(self.ha.run_cycle(), 'starting as a secondary') + self.p.is_running = true + self.ha.follow = Mock(return_value='fake') + self.assertEqual(self.ha.run_cycle(), 'fake') @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)) @patch.object(Bootstrap, 'create_replica', Mock(return_value=1)) diff --git a/tests/test_raft.py b/tests/test_raft.py index 716295da..a80a668d 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -4,7 +4,7 @@ import tempfile import time from mock import Mock, PropertyMock, patch -from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility +from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility, TCPTransport, _TCPTransport from pysyncobj import SyncObjConf, FAIL_REASON @@ -23,6 +23,16 @@ def remove_files(prefix): time.sleep(1.0) +class TestTCPTransport(unittest.TestCase): + + @patch.object(TCPTransport, '__init__', Mock()) + @patch.object(TCPTransport, 'setOnUtilityMessageCallback', Mock()) + @patch.object(TCPTransport, '_connectIfNecessarySingle', Mock(side_effect=Exception)) + def test__connectIfNecessarySingle(self): + t = _TCPTransport(Mock(), None, []) + self.assertFalse(t._connectIfNecessarySingle(None)) + + @patch('pysyncobj.tcp_server.TcpServer.bind', Mock()) class TestDynMemberSyncObj(unittest.TestCase): @@ -114,8 +124,7 @@ class TestRaft(unittest.TestCase): def test_raft(self): raft = Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'self_addr': '127.0.0.1:1234', 'retry_timeout': 10, 'data_dir': self._TMP}) - raft.set_retry_timeout(20) - raft.set_ttl(60) + raft.reload_config({'retry_timeout': 20, 'ttl': 60, 'loop_wait': 10}) self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}')) self.assertTrue(raft.touch_member('')) self.assertTrue(raft.initialize())