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())