Fix issues with raft traffic encryption (#1919)

and run raft behave tests with encryption enabled.

Using the new `pysyncobj` release allowed us to get rid of a lot of hacks with accessing private properties and methods of the parent class and reduce the size of the `raft.py`.

Close https://github.com/zalando/patroni/issues/1746

.github/workflows/install_deps.py

@@ -22,7 +22,7 @@ def install_requirements(what):
     for r in read('requirements.txt').split('\n'):
         r = r.strip()
         if r != '':
-            extras = {e for e, v in EXTRAS_REQUIRE.items() if v and r.startswith(v[0])}
+            extras = {e for e, v in EXTRAS_REQUIRE.items() if v and any(r.startswith(x) for x in v)}
             if not extras or what == 'all' or what in extras:
                 requirements.append(r)
 

features/environment.py

@@ -591,10 +591,12 @@ class ExhibitorController(ZooKeeperController):
 class RaftController(AbstractDcsController):
 
     CONTROLLER_ADDR = 'localhost:1234'
+    PASSWORD = '12345'
 
     def __init__(self, context):
         super(RaftController, self).__init__(context)
-        os.environ.update(PATRONI_RAFT_PARTNER_ADDRS="'" + self.CONTROLLER_ADDR + "'", RAFT_PORT='1234')
+        os.environ.update(PATRONI_RAFT_PARTNER_ADDRS="'" + self.CONTROLLER_ADDR + "'",
+                          PATRONI_RAFT_PASSWORD=self.PASSWORD, RAFT_PORT='1234')
         self._raft = None
 
     def _start(self):
@@ -615,18 +617,16 @@ class RaftController(AbstractDcsController):
 
     def cleanup_service_tree(self):
         from patroni.dcs.raft import KVStoreTTL
-        from pysyncobj import SyncObjConf
 
         if self._raft:
             self._raft.destroy()
-            self._raft._SyncObj__thread.join()
             self.stop()
             os.makedirs(self._work_directory)
             self.start()
 
         ready_event = threading.Event()
-        conf = SyncObjConf(appendEntriesUseBatch=False, dynamicMembershipChange=True, onReady=ready_event.set)
-        self._raft = KVStoreTTL(None, [self.CONTROLLER_ADDR], conf)
+        self._raft = KVStoreTTL(ready_event.set, None, None, partner_addrs=[self.CONTROLLER_ADDR], password=self.PASSWORD)
+        self._raft.startAutoTick()
         ready_event.wait()
 
 

patroni/dcs/raft.py

@@ -5,7 +5,8 @@ import threading
 import time
 
 from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON
-from pysyncobj.transport import Node, TCPTransport, CONNECTION_STATE
+from pysyncobj.transport import TCPTransport, CONNECTION_STATE
+from pysyncobj.utility import TcpUtility, UtilityException
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
 from ..utils import validate_directory
@@ -13,122 +14,97 @@ from ..utils import validate_directory
 logger = logging.getLogger(__name__)
 
 
-class MessageNode(Node):
-
-    def __init__(self, address):
-        self.address = address
-
-
-class UtilityTransport(TCPTransport):
+class _TCPTransport(TCPTransport):
 
     def __init__(self, syncObj, selfNode, otherNodes):
-        super(UtilityTransport, self).__init__(syncObj, selfNode, otherNodes)
-        self._selfIsReadonlyNode = False
-
-    def _connectIfNecessarySingle(self, node):
-        pass
-
-    def connectionState(self, node):
-        return self._connections[node].state
-
-    def isDisconnected(self, node):
-        return self.connectionState(node) == CONNECTION_STATE.DISCONNECTED
-
-    def connectIfRequiredSingle(self, node):
-        if self.isDisconnected(node):
-            return self._connections[node].connect(node.ip, node.port)
-
-    def disconnectSingle(self, node):
-        self._connections[node].disconnect()
+        super(_TCPTransport, self).__init__(syncObj, selfNode, otherNodes)
+        self.setOnUtilityMessageCallback('members', syncObj.getMembers)
 
 
-class SyncObjUtility(SyncObj):
+class SyncObjUtility(object):
 
     def __init__(self, otherNodes, conf):
-        autoTick = conf.autoTick
-        conf.autoTick = False
-        super(SyncObjUtility, self).__init__(None, otherNodes, conf, transportClass=UtilityTransport)
-        conf.autoTick = autoTick
-        self._SyncObj__transport.setOnMessageReceivedCallback(self._onMessageReceived)
-        self.__result = None
+        self._nodes = otherNodes
+        self._utility = TcpUtility(conf.password)
 
-    def setPartnerNode(self, partner):
-        self.__node = partner
+    def executeCommand(self, command):
+        try:
+            return self._utility.executeCommand(self.__node, command)
+        except UtilityException:
+            return None
 
-    def sendMessage(self, message):
-        # Abuse the fact that node address is send as a first message
-        self._SyncObj__transport._selfNode = MessageNode(message)
-        self._SyncObj__transport.connectIfRequiredSingle(self.__node)
-        while not self._SyncObj__transport.isDisconnected(self.__node):
-            self._poller.poll(0.5)
-        return self.__result
-
-    def _onMessageReceived(self, _, message):
-        self.__result = message
-        self._SyncObj__transport.disconnectSingle(self.__node)
-
-
-class MyTCPTransport(TCPTransport):
-
-    def _onIncomingMessageReceived(self, conn, message):
-        if self._syncObj.encryptor and not conn.sendRandKey:
-            conn.sendRandKey = message
-            conn.recvRandKey = os.urandom(32)
-            conn.send(conn.recvRandKey)
-            return
-
-        # Utility messages
-        if isinstance(message, list) and message[0] == 'members':
-            conn.send(self._syncObj._get_members())
-            return True
-
-        return super(MyTCPTransport, self)._onIncomingMessageReceived(conn, message)
+    def getMembers(self):
+        for self.__node in self._nodes:
+            response = self.executeCommand(['members'])
+            if response:
+                return [member['addr'] for member in response]
 
 
 class DynMemberSyncObj(SyncObj):
 
     def __init__(self, selfAddress, partnerAddrs, conf):
-        add_self = False
+        self.__early_apply_local_log = selfAddress is not None
+        self.applied_local_log = False
+
         utility = SyncObjUtility(partnerAddrs, conf)
-        for node in utility._SyncObj__otherNodes:
-            utility.setPartnerNode(node)
-            response = utility.sendMessage(['members'])
-            if response:
-                partnerAddrs = [member['addr'] for member in response if member['addr'] != selfAddress]
-                add_self = selfAddress and len(partnerAddrs) == len(response)
-                break
+        members = utility.getMembers()
+        add_self = members and selfAddress not in members
+
+        partnerAddrs = [member for member in (members or partnerAddrs) if member != selfAddress]
+
+        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=_TCPTransport)
 
-        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=MyTCPTransport)
         if add_self:
-            threading.Thread(target=utility.sendMessage, args=(['add', selfAddress],)).start()
+            thread = threading.Thread(target=utility.executeCommand, args=(['add', selfAddress],))
+            thread.daemon = True
+            thread.start()
 
-    def _get_members(self):
-        ret = [{'addr': node.id, 'leader': node == self._getLeader(),
-                'status': CONNECTION_STATE.CONNECTED if node in self._SyncObj__connectedNodes
-                else CONNECTION_STATE.DISCONNECTED} for node in self._SyncObj__otherNodes]
-        ret.append({'addr': self._SyncObj__selfNode.id, 'leader': self._isLeader(),
-                    'status': CONNECTION_STATE.CONNECTED})
-        return ret
+    def getMembers(self, args, callback):
+        callback([{'addr': node.id, 'leader': node == self._getLeader(), 'status': CONNECTION_STATE.CONNECTED
+                   if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] +
+                 [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None)
 
-    def _SyncObj__doChangeCluster(self, request, reverse=False):
-        ret = False
-        if not self._SyncObj__selfNode or request[0] != 'add' or reverse or request[1] != self._SyncObj__selfNode.id:
-            ret = super(DynMemberSyncObj, self)._SyncObj__doChangeCluster(request, reverse)
-            if ret:
-                self.forceLogCompaction()
-        return ret
+    def _onTick(self, timeToWait=0.0):
+        super(DynMemberSyncObj, self)._onTick(timeToWait)
+
+        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
+        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
+        # We are using the `applied_local_log` property for that, but not calling the callback function.
+        if self.__early_apply_local_log and not self.applied_local_log and self.raftLastApplied == self.raftCommitIndex:
+            self.applied_local_log = True
 
 
 class KVStoreTTL(DynMemberSyncObj):
 
-    def __init__(self, selfAddress, partnerAddrs, conf, on_set=None, on_delete=None):
+    def __init__(self, on_ready, on_set, on_delete, **config):
+        self.__thread = None
         self.__on_set = on_set
         self.__on_delete = on_delete
         self.__limb = {}
         self.__retry_timeout = None
-        self.__early_apply_local_log = selfAddress is not None
-        self.applied_local_log = False
-        super(KVStoreTTL, self).__init__(selfAddress, partnerAddrs, conf)
+
+        self_addr = config.get('self_addr')
+        partner_addrs = set(config.get('partner_addrs', []))
+        if config.get('patronictl'):
+            if self_addr:
+                partner_addrs.add(self_addr)
+            self_addr = None
+
+        # Create raft data_dir if necessary
+        raft_data_dir = config.get('data_dir', '')
+        if raft_data_dir != '':
+            validate_directory(raft_data_dir)
+
+        file_template = (self_addr or '')
+        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'),
+                           fullDumpFile=(file_template + '.dump' if self_addr else None),
+                           journalFile=(file_template + '.journal' if self_addr else None),
+                           onReady=on_ready, dynamicMembershipChange=True)
+
+        super(KVStoreTTL, self).__init__(self_addr, partner_addrs, conf)
         self.__data = {}
 
     @staticmethod
@@ -175,7 +151,7 @@ class KVStoreTTL(DynMemberSyncObj):
 
         if old_value and old_value['created'] != value['created']:
             value['created'] = value['updated']
-        value['index'] = self._SyncObj__raftLastApplied + 1
+        value['index'] = self.raftLastApplied + 1
 
         self.__data[key] = value
         if self.__on_set:
@@ -242,27 +218,29 @@ class KVStoreTTL(DynMemberSyncObj):
         return {k: v for k, v in self.__data.items() if k.startswith(key)}
 
     def _onTick(self, timeToWait=0.0):
-        # The SyncObj starts applying the local log only when there is at least one node connected.
-        # We want to change this behavior and apply the local log even when there is nobody except us.
-        # It gives us at least some picture about the last known cluster state.
-        if self.__early_apply_local_log and not self.applied_local_log and self._SyncObj__needLoadDumpFile:
-            self._SyncObj__raftCommitIndex = self._SyncObj__getCurrentLogIndex()
-            self._SyncObj__raftCurrentTerm = self._SyncObj__getCurrentLogTerm()
-
         super(KVStoreTTL, self)._onTick(timeToWait)
 
-        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
-        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
-        # We are using the `applied_local_log` property for that, but not calling the callback function.
-        if self.__early_apply_local_log and not self.applied_local_log and self._SyncObj__raftCommitIndex != 1 and \
-                self._SyncObj__raftLastApplied == self._SyncObj__raftCommitIndex:
-            self.applied_local_log = True
-
         if self._isLeader():
             self.__expire_keys()
         else:
             self.__limb.clear()
 
+    def _autoTickThread(self):
+        self.__destroying = False
+        while not self.__destroying:
+            self.doTick(self.conf.autoTickPeriod)
+
+    def startAutoTick(self):
+        self.__thread = threading.Thread(target=self._autoTickThread)
+        self.__thread.daemon = True
+        self.__thread.start()
+
+    def destroy(self):
+        if self.__thread:
+            self.__destroying = True
+            self.__thread.join()
+        super(KVStoreTTL, self).destroy()
+
 
 class Raft(AbstractDCS):
 
@@ -270,34 +248,16 @@ class Raft(AbstractDCS):
         super(Raft, self).__init__(config)
         self._ttl = int(config.get('ttl') or 30)
 
-        self_addr = config.get('self_addr')
-        partner_addrs = config.get('partner_addrs', [])
-        if self._ctl:
-            if self_addr:
-                partner_addrs.append(self_addr)
-            self_addr = None
-
-        # Create raft data_dir if necessary
-        raft_data_dir = config.get('data_dir', '')
-        if raft_data_dir != '':
-            validate_directory(raft_data_dir)
-
         ready_event = threading.Event()
-        file_template = os.path.join(config.get('data_dir', ''), (self_addr or ''))
-        conf = SyncObjConf(password=config.get('password'), appendEntriesUseBatch=False,
-                           bindAddress=config.get('bind_addr'), commandsWaitLeader=False,
-                           fullDumpFile=(file_template + '.dump' if self_addr else None),
-                           journalFile=(file_template + '.journal' if self_addr else None),
-                           onReady=ready_event.set, dynamicMembershipChange=True)
+        self._sync_obj = KVStoreTTL(ready_event.set, self._on_set, self._on_delete, commandsWaitLeader=False, **config)
+        self._sync_obj.startAutoTick()
 
-        self._sync_obj = KVStoreTTL(self_addr, partner_addrs, conf, self._on_set, self._on_delete)
         while True:
             ready_event.wait(5)
             if ready_event.isSet() or self._sync_obj.applied_local_log:
                 break
             else:
                 logger.info('waiting on raft')
-        self._sync_obj.forceLogCompaction()
         self.set_retry_timeout(int(config.get('retry_timeout') or 10))
 
     def _on_set(self, key, value):

patroni/raft_controller.py

@@ -1,9 +1,7 @@
 import logging
-import os
 
-from patroni.daemon import AbstractPatroniDaemon, abstract_main
-from patroni.dcs.raft import KVStoreTTL
-from pysyncobj import SyncObjConf
+from .daemon import AbstractPatroniDaemon, abstract_main
+from .dcs.raft import KVStoreTTL
 
 logger = logging.getLogger(__name__)
 
@@ -13,16 +11,13 @@ class RaftController(AbstractPatroniDaemon):
     def __init__(self, config):
         super(RaftController, self).__init__(config)
 
-        raft_config = self.config.get('raft')
-        self_addr = raft_config['self_addr']
-        template = os.path.join(raft_config.get('data_dir', ''), self_addr)
-        self._syncobj_config = SyncObjConf(autoTick=False, appendEntriesUseBatch=False, dynamicMembershipChange=True,
-                                           journalFile=template + '.journal', fullDumpFile=template + '.dump')
-        self._raft = KVStoreTTL(self_addr, raft_config.get('partner_addrs', []), self._syncobj_config)
+        config = self.config.get('raft')
+        assert 'self_addr' in config
+        self._raft = KVStoreTTL(None, None, None, **config)
 
     def _run_cycle(self):
         try:
-            self._raft.doTick(self._syncobj_config.autoTickPeriod)
+            self._raft.doTick(self._raft.conf.autoTickPeriod)
         except Exception:
             logger.exception('doTick')
 

requirements.txt

@@ -9,6 +9,7 @@ python-consul>=0.7.1
 click>=4.1
 prettytable>=0.7
 python-dateutil
-pysyncobj>=0.3.7
+pysyncobj>=0.3.8
+cryptography>=1.4
 psutil>=2.0.0
 ydiff>=1.2.0

setup.py

@@ -22,8 +22,9 @@ AUTHOR_EMAIL = '[email protected], [email protected], alexk
 KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
     ' zookeeper exhibitor consul streaming replication kubernetes k8s'
 
-EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'], 'consul': ['python-consul'],
-                  'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['ipaddress'], 'raft': ['pysyncobj']}
+EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'],
+                  'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'],
+                  'kubernetes': ['ipaddress'], 'raft': ['pysyncobj', 'cryptography']}
 COVERAGE_XML = True
 COVERAGE_HTML = False
 
@@ -171,10 +172,16 @@ def setup_package(version):
         if r == '':
             continue
         extra = False
-        for e, v in EXTRAS_REQUIRE.items():
-            if v and r.startswith(v[0]):
-                EXTRAS_REQUIRE[e] = [r] if e != 'kubernetes' or sys.version_info < (3, 0, 0) else []
+        for e, deps in EXTRAS_REQUIRE.items():
+            for i, v in enumerate(deps):
+                if r.startswith(v):
+                    if e != 'kubernetes' or sys.version_info < (3, 0, 0):
+                        deps[i] = r
+                    else:
+                        deps = []
+                    EXTRAS_REQUIRE[e] = deps
                     extra = True
+                    break
         if not extra:
             install_requires.append(r)
 

tests/test_raft.py

@@ -3,13 +3,13 @@ import unittest
 import tempfile
 import time
 
-from mock import Mock, patch
+from mock import Mock, PropertyMock, patch
 from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility
 from pysyncobj import SyncObjConf, FAIL_REASON
 
 
 def remove_files(prefix):
-    for f in ('journal', 'dump'):
+    for f in ('journal', 'journal.meta', 'dump'):
         f = prefix + f
         if os.path.isfile(f):
             for i in range(0, 15):
@@ -31,50 +31,38 @@ class TestDynMemberSyncObj(unittest.TestCase):
         self.conf = SyncObjConf(appendEntriesUseBatch=False, dynamicMembershipChange=True, autoTick=False)
         self.so = DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
 
-    @patch.object(SyncObjUtility, 'sendMessage')
-    def test_add_member(self, mock_send_message):
-        mock_send_message.return_value = [{'addr': '127.0.0.1:1235'}, {'addr': '127.0.0.1:1236'}]
-        mock_send_message.ver = 0
+    @patch.object(SyncObjUtility, 'executeCommand')
+    def test_add_member(self, mock_execute_command):
+        mock_execute_command.return_value = [{'addr': '127.0.0.1:1235'}, {'addr': '127.0.0.1:1236'}]
+        mock_execute_command.ver = 0
         DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
         self.conf.dynamicMembershipChange = False
         DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
 
-    def test___onUtilityMessage(self):
-        self.so._SyncObj__encryptor = Mock()
+    def test_getMembers(self):
         mock_conn = Mock()
-        mock_conn.sendRandKey = None
-        self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, 'randkey')
         self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, ['members'])
-        self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, ['status'])
 
     def test__SyncObj__doChangeCluster(self):
         self.so._SyncObj__doChangeCluster(['add', '127.0.0.1:1236'])
 
-    def test_utility(self):
-        utility = SyncObjUtility(['127.0.0.1:1235'], self.conf)
-        utility.setPartnerNode(list(utility._SyncObj__otherNodes)[0])
-        utility.sendMessage(['members'])
-        utility._onMessageReceived(0, '')
-
 
+@patch.object(SyncObjConf, 'fullDumpFile', PropertyMock(return_value=None), create=True)
+@patch.object(SyncObjConf, 'journalFile', PropertyMock(return_value=None), create=True)
 class TestKVStoreTTL(unittest.TestCase):
 
+    @patch.object(SyncObjConf, 'fullDumpFile', PropertyMock(return_value=None), create=True)
+    @patch.object(SyncObjConf, 'journalFile', PropertyMock(return_value=None), create=True)
     def setUp(self):
-        self.conf = SyncObjConf(appendEntriesUseBatch=False, appendEntriesPeriod=0.001,
-                                raftMinTimeout=0.004, raftMaxTimeout=0.005, autoTickPeriod=0.001)
         callback = Mock()
         callback.replicated = False
-        self.so = KVStoreTTL('127.0.0.1:1234', [], self.conf, on_set=callback, on_delete=callback)
+        self.so = KVStoreTTL(None, callback, callback, self_addr='127.0.0.1:1234')
+        self.so.startAutoTick()
         self.so.set_retry_timeout(10)
 
-    @staticmethod
-    def destroy(so):
-        so.destroy()
-        so._SyncObj__thread.join()
-
     def tearDown(self):
         if self.so:
-            self.destroy(self.so)
+            self.so.destroy()
 
     def test_set(self):
         self.assertTrue(self.so.set('foo', 'bar', prevExist=False, ttl=30))
@@ -83,7 +71,7 @@ class TestKVStoreTTL(unittest.TestCase):
         self.assertTrue(self.so.retry(self.so._set, 'foo', {'value': 'buz', 'created': 1, 'updated': 1}))
 
     def test_delete(self):
-        self.conf.autoTickPeriod = 0.1
+        self.so.autoTickPeriod = 0.2
         self.so.set('foo', 'bar')
         self.so.set('fooo', 'bar')
         self.assertFalse(self.so.delete('foo', prevValue='buz'))
@@ -111,11 +99,10 @@ class TestKVStoreTTL(unittest.TestCase):
 
     def test_on_ready_override(self):
         self.assertTrue(self.so.set('foo', 'bar'))
-        self.destroy(self.so)
+        self.so.destroy()
         self.so = None
-        self.conf.onReady = Mock()
-        self.conf.autoTick = False
-        so = KVStoreTTL('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
+        so = KVStoreTTL(Mock(), None, None, self_addr='127.0.0.1:1234',
+                        partner_addrs=['127.0.0.1:1235'], patronictl=True)
         so.doTick(0)
         so.destroy()
 
@@ -150,7 +137,6 @@ class TestRaft(unittest.TestCase):
         self.assertTrue(raft.take_leader())
         raft.watch(None, 0.001)
         raft._sync_obj.destroy()
-        raft._sync_obj._SyncObj__thread.join()
 
     def tearDown(self):
         remove_files(os.path.join(self._TMP, '127.0.0.1:1234.'))