diff --git a/patroni/dcs.py b/patroni/dcs.py index 1bca092e..d007b392 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -3,8 +3,8 @@ import json from collections import namedtuple from patroni.exceptions import DCSError -from patroni.utils import sleep from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl +from threading import Event def parse_connection_string(value): @@ -42,12 +42,17 @@ class Member(namedtuple('Member', 'index,name,session,data')): """ >>> Member.from_node(-1, '', '', '{"conn_url": "postgres://foo@bar/postgres"}') is not None True + >>> Member.from_node(-1, '', '', '{') + Member(index=-1, name='', session='', data={}) """ if data.startswith('postgres'): conn_url, api_url = parse_connection_string(data) data = {'conn_url': conn_url, 'api_url': api_url} else: - data = json.loads(data) + try: + data = json.loads(data) + except: + data = {} return Member(index, name, session, data) @property @@ -121,6 +126,7 @@ class AbstractDCS: self._base_path = '/service/' + self._scope self.cluster = None + self.event = Event() def client_path(self, path): return '/'.join([self._base_path, path.lstrip('/')]) @@ -234,5 +240,5 @@ class AbstractDCS: :param timeout: timeout in seconds :returns: `!True` if you would like to reschedule the next run of ha cycle""" - sleep(timeout) - return False + self.event.wait(timeout) + return self.event.isSet() diff --git a/patroni/etcd.py b/patroni/etcd.py index f1ca5044..76e00554 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -147,7 +147,6 @@ class Etcd(AbstractDCS): def __init__(self, name, config): super(Etcd, self).__init__(name, config) self.ttl = config['ttl'] - self.member_ttl = config.get('member_ttl', 3600) self._retry = Retry(deadline=10, max_delay=1, max_tries=-1, retry_exceptions=(etcd.EtcdConnectionFailed, etcd.EtcdLeaderElectionInProgress, @@ -210,7 +209,7 @@ class Etcd(AbstractDCS): @catch_etcd_errors def touch_member(self, connection_string, ttl=None): - return self.retry(self.client.set, self.member_path, connection_string, ttl or self.member_ttl) + return self.retry(self.client.set, self.member_path, connection_string, ttl or self.ttl) @catch_etcd_errors def take_leader(self): @@ -269,4 +268,7 @@ class Etcd(AbstractDCS): timeout = end_time - time.time() - return timeout > 0 and super(Etcd, self).watch(timeout) + try: + return super(Etcd, self).watch(timeout) + finally: + self.event.clear() diff --git a/patroni/ha.py b/patroni/ha.py index 25a05c23..ca79d8e3 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -150,7 +150,8 @@ class Ha: if self.state_handler.is_leader(): return True - if check_replication_lag and not self.state_handler.check_replication_lag(self.dcs.cluster.last_leader_operation): + if check_replication_lag and \ + not self.state_handler.check_replication_lag(self.dcs.cluster.last_leader_operation): return False # Too far behind last reported xlog location on master # Prepare list of nodes to run check against diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index a5258eac..68786485 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -91,7 +91,6 @@ class ZooKeeper(AbstractDCS): 'max_tries': -1}, connection_retry={'max_delay': 1, 'max_tries': -1}) self.client.add_listener(self.session_listener) - self.cluster_event = self.client.handler.event_object() self._my_member_data = None self.fetch_cluster = True @@ -105,7 +104,7 @@ class ZooKeeper(AbstractDCS): def cluster_watcher(self, event): self.fetch_cluster = True - self.cluster_event.set() + self.event.set() def get_node(self, key, watch=None): try: @@ -133,7 +132,7 @@ class ZooKeeper(AbstractDCS): return members def _inner_load_cluster(self): - self.cluster_event.clear() + self.event.clear() nodes = set(self.get_children(self.client_path(''), self.cluster_watcher)) # get initialize flag @@ -279,8 +278,5 @@ class ZooKeeper(AbstractDCS): logger.exception("Unable to delete initialize key") def watch(self, timeout): - self.cluster_event.wait(timeout) - if self.cluster_event.isSet(): - self.fetch_cluster = True - return True - return False + self.fetch_cluster = super(ZooKeeper, self).watch(timeout) + return self.fetch_cluster diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 3d2535a7..53c054e5 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -1,4 +1,3 @@ -import datetime import etcd import json import requests @@ -8,7 +7,7 @@ import unittest from dns.exception import DNSException from mock import Mock, patch -from patroni.dcs import Cluster, DCSError, Leader, Member +from patroni.dcs import Cluster, DCSError, Leader from patroni.etcd import Client, Etcd @@ -194,7 +193,6 @@ class TestClient(unittest.TestCase): self.assertRaises(etcd.EtcdException, self.client._load_machines_cache) -@patch('time.sleep', Mock()) @patch('requests.get', requests_get) class TestEtcd(unittest.TestCase): @@ -254,7 +252,7 @@ class TestEtcd(unittest.TestCase): def test_watch(self): self.etcd.client.watch = etcd_watch - self.etcd.watch(100) + self.etcd.watch(0) self.etcd.get_cluster() self.etcd.watch(1.5) self.etcd.watch(4.5) diff --git a/tests/test_ha.py b/tests/test_ha.py index a0a43d3f..392ad663 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -6,7 +6,6 @@ from patroni.etcd import Client, Etcd from patroni.exceptions import DCSError, PostgresException from patroni.ha import Ha from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get -from threading import Thread def true(*args, **kwargs): @@ -84,6 +83,7 @@ class MockPatroni: self.api = Mock() self.api.connection_string = 'http://127.0.0.1:8008' + def run_async(func, args=()): func(args) if args else func() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index b38f36ce..52b10d7a 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -1,4 +1,3 @@ -import datetime import sys import time import unittest @@ -7,9 +6,7 @@ import yaml from mock import Mock, patch from patroni.api import RestApiServer from patroni.async_executor import AsyncExecutor -from patroni.dcs import Cluster, Member from patroni.etcd import Etcd -from patroni.ha import Ha from patroni import Patroni, main from patroni.zookeeper import ZooKeeper from six.moves import BaseHTTPServer diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 9e14bec2..b4e004be 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -170,5 +170,5 @@ class TestZooKeeper(unittest.TestCase): def test_watch(self): self.zk.watch(0) - self.zk.cluster_event.isSet = lambda: False + self.zk.event.isSet = lambda: False self.zk.watch(0)