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

@@ -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'])

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

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

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