diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py index 1a4477ea..fb5194cc 100644 --- a/.github/workflows/install_deps.py +++ b/.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) diff --git a/features/environment.py b/features/environment.py index d457ae12..130a5bf5 100644 --- a/features/environment.py +++ b/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() diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index f94e6e64..76c65efb 100644 --- a/patroni/dcs/raft.py +++ b/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): diff --git a/patroni/raft_controller.py b/patroni/raft_controller.py index 5a9ae704..1baa9786 100644 --- a/patroni/raft_controller.py +++ b/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') diff --git a/requirements.txt b/requirements.txt index 9fc6f4b4..c7438d3d 100644 --- a/requirements.txt +++ b/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 diff --git a/setup.py b/setup.py index 17dd696b..7302858a 100644 --- a/setup.py +++ b/setup.py @@ -22,8 +22,9 @@ AUTHOR_EMAIL = 'alexander.kukushkin@zalando.de, dmitrii.dolgov@zalando.de, 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 [] - extra = True + 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) diff --git a/tests/test_raft.py b/tests/test_raft.py index 54197df8..716295da 100644 --- a/tests/test_raft.py +++ b/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.'))