use Event.wait instead of sleep

it makes possible to break "sleep" for example from API

plus small bugfix: catch ValueError exception from json.loads

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

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

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

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

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)

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

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

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)