refactoring:

1. run touch_member from the main loop
2. move code which takes care about long tasks into separate class
3. change format of data stored in a DCS: use json instead of url
4. change Member class: from now it deserialize everything into data property
5. rework API: from now it takes into account state of the current node in a dcs

patroni/__init__.py

@@ -19,11 +19,11 @@ class Patroni:
     def __init__(self, config):
         self.nap_time = config['loop_wait']
         self.postgresql = Postgresql(config['postgresql'])
-        self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config))
+        self.dcs = self.get_dcs(self.postgresql.name, config)
         host, port = config['restapi']['listen'].split(':')
         self.api = RestApiServer(self, config['restapi'])
+        self.ha = Ha(self)
         self.next_run = time.time()
-        self.shutdown_member_ttl = 300
 
     @staticmethod
     def get_dcs(name, config):
@@ -33,22 +33,13 @@ class Patroni:
             return ZooKeeper(name, config['zookeeper'])
         raise Exception('Can not find sutable configuration of distributed configuration store')
 
-    def touch_member(self, ttl=None):
-        connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
-        if self.ha.cluster:
-            for m in self.ha.cluster.members:
-                # Do not update member TTL when it is far from being expired
-                if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl:
-                    return True
-        return self.ha.dcs.touch_member(connection_string, ttl)
-
     def schedule_next_run(self):
         self.next_run += self.nap_time
         current_time = time.time()
         nap_time = self.next_run - current_time
         if nap_time <= 0:
             self.next_run = current_time
-        elif self.ha.dcs.watch(nap_time):
+        elif self.dcs.watch(nap_time):
             self.next_run = time.time()
 
     def run(self):
@@ -56,7 +47,6 @@ class Patroni:
         self.next_run = time.time()
 
         while True:
-            self.touch_member()
             logger.info(self.ha.run_cycle())
             reap_children()
             self.schedule_next_run()
@@ -81,6 +71,5 @@ def main():
         pass
     finally:
         patroni.api.shutdown()
-        patroni.touch_member(patroni.shutdown_member_ttl)  # schedule member removal
         patroni.postgresql.stop()
-        patroni.ha.dcs.delete_leader()
+        patroni.dcs.delete_leader()

patroni/api.py

@@ -48,7 +48,18 @@ class RestApiHandler(BaseHTTPRequestHandler):
         response = self.get_postgresql_status()
 
         patroni = self.server.patroni
-        if 'role' in response and response['role'] in path:
+        if patroni.dcs.cluster:  # dcs available
+            if patroni.dcs.cluster.leader and patroni.dcs.cluster.leader.name == patroni.postgresql.name:  # is_leader
+                status_code = 200 if 'master' in path else 503
+            elif 'role' not in response:
+                status_code = 503
+            elif response['role'] == 'master':  # running as master but without leader lock!!!!
+                status_code = 503
+            elif response['role'] in path:
+                status_code = 200
+            else:
+                status_code = 503
+        elif 'role' in response and response['role'] in path:
             status_code = 200
         elif patroni.ha.restart_scheduled() and patroni.postgresql.role == 'master' and 'master' in path:
             # exceptional case for master node when the postgres is being restarted via API

patroni/async_executor.py

@@ -0,0 +1,55 @@
+import logging
+from threading import Lock, Thread
+
+logger = logging.getLogger(__name__)
+
+
+class AsyncExecutor:
+
+    def __init__(self):
+        Lock.__init__(self)
+        self._busy = False
+        self._thread_lock = Lock()
+        self._scheduled_action = None
+        self._scheduled_action_lock = Lock()
+
+    @property
+    def busy(self):
+        return self._busy
+
+    def schedule(self, action, immediately=False):
+        with self._scheduled_action_lock:
+            if self._scheduled_action is not None:
+                return self._scheduled_action
+            self._scheduled_action = action
+            self._busy = immediately
+        return None
+
+    @property
+    def scheduled_action(self):
+        with self._scheduled_action_lock:
+            return self._scheduled_action
+
+    def reset_scheduled_action(self):
+        with self._scheduled_action_lock:
+            self._scheduled_action = None
+
+    def run(self, func, args=()):
+        try:
+            return func(*args) if args else func()
+        except:
+            logger.exception('Exception during execution of long running task %s', self.scheduled_action)
+        finally:
+            with self:
+                self._busy = False
+                self.reset_scheduled_action()
+
+    def run_async(self, func, args=()):
+        self._busy = True
+        Thread(target=self.run, args=(func, args)).start()
+
+    def __enter__(self):
+        self._thread_lock.acquire()
+
+    def __exit__(self, type, value, traceback):
+        self._thread_lock.release()

patroni/dcs.py

@@ -1,8 +1,9 @@
 import abc
+import json
 
 from collections import namedtuple
 from patroni.exceptions import DCSError
-from patroni.utils import calculate_ttl, sleep
+from patroni.utils import sleep
 from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
 
 
@@ -23,28 +24,47 @@ def parse_connection_string(value):
     return conn_url, api_url
 
 
-class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')):
+class Member(namedtuple('Member', 'index,name,session,data')):
 
     """Immutable object (namedtuple) which represents single member of PostgreSQL cluster.
     Consists of the following fields:
     :param index: modification index of a given member key in a Configuration Store
     :param name: name of PostgreSQL cluster member
-    :param conn_url: connection string containing host, user and password which could be used to access this member.
-    :param api_url: REST API url of patroni instance
-    :param expiration: expiration time of given member key
-    :param ttl: ttl of given member key in seconds"""
+    :param session: either session id or just ttl in seconds
+    :param data: arbitrary data i.e. conn_url, api_url, xlog location, state, role, tags, etc...
 
-    def real_ttl(self):
-        return calculate_ttl(self.expiration) or -1
+    There are two mandatory keys in a data:
+    conn_url: connection string containing host, user and password which could be used to access this member.
+    api_url: REST API url of patroni instance"""
+
+    @staticmethod
+    def from_node(index, name, session, data):
+        """
+        >>> Member.from_node(-1, '', '', '{"conn_url": "postgres://foo@bar/postgres"}') is not None
+        True
+        """
+        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)
+        return Member(index, name, session, data)
+
+    @property
+    def conn_url(self):
+        return self.data.get('conn_url', None)
+
+    @property
+    def api_url(self):
+        return self.data.get('api_url', None)
 
 
-class Leader(namedtuple('Leader', 'index,expiration,ttl,member')):
+class Leader(namedtuple('Leader', 'index,session,member')):
 
     """Immutable object (namedtuple) which represents leader key.
     Consists of the following fields:
     :param index: modification index of a leader key in a Configuration Store
-    :param expiration: expiration time of the leader key
-    :param ttl: ttl of the leader key
+    :param session: either session id or just ttl in seconds
     :param member: reference to a `Member` object which represents current leader (see `Cluster.members`)"""
 
     @property
@@ -100,6 +120,8 @@ class AbstractDCS:
         self._scope = config['scope']
         self._base_path = '/service/' + self._scope
 
+        self.cluster = None
+
     def client_path(self, path):
         return '/'.join([self._base_path, path.lstrip('/')])
 

patroni/etcd.py

@@ -10,7 +10,7 @@ import urllib3
 
 from dns.exception import DNSException
 from dns import resolver
-from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member, parse_connection_string
+from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member
 from patroni.exceptions import DCSError
 from patroni.utils import Retry, RetryFailedError, sleep
 from requests.exceptions import RequestException
@@ -154,7 +154,6 @@ class Etcd(AbstractDCS):
                                               etcd.EtcdWatcherCleared,
                                               etcd.EtcdEventIndexCleared))
         self.client = self.get_etcd_client(config)
-        self.cluster = None
 
     def retry(self, *args, **kwargs):
         return self._retry.copy()(*args, **kwargs)
@@ -171,8 +170,7 @@ class Etcd(AbstractDCS):
 
     @staticmethod
     def member(node):
-        conn_url, api_url = parse_connection_string(node.value)
-        return Member(node.modifiedIndex, os.path.basename(node.key), conn_url, api_url, node.expiration, node.ttl)
+        return Member.from_node(node.modifiedIndex, os.path.basename(node.key), node.ttl, node.value)
 
     def get_cluster(self):
         try:
@@ -192,9 +190,9 @@ class Etcd(AbstractDCS):
             # get leader
             leader = nodes.get(self._LEADER, None)
             if leader:
-                member = Member(-1, leader.value, None, None, None, None)
+                member = Member(-1, leader.value, None, {})
                 member = ([m for m in members if m.name == leader.value] or [member])[0]
-                leader = Leader(leader.modifiedIndex, leader.expiration, leader.ttl, member)
+                leader = Leader(leader.modifiedIndex, leader.ttl, member)
 
             # failover key
             failover = nodes.get(self._FAILOVER, None)

patroni/ha.py

@@ -1,35 +1,28 @@
+import json
 import logging
 import psycopg2
 import requests
 
+from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
 from multiprocessing.pool import ThreadPool
-from threading import Lock, Thread
 
 logger = logging.getLogger(__name__)
 
 
 class Ha:
 
-    def __init__(self, state_handler, dcs):
-        self.state_handler = state_handler
-        self.dcs = dcs
-        self.cluster = None
+    def __init__(self, patroni):
+        self.patroni = patroni
+        self.state_handler = patroni.postgresql
+        self.dcs = patroni.dcs
         self.old_cluster = None
-        self._scheduled_action = None
-        self._scheduled_action_lock = Lock()
-        self._long_action_in_progress = False
-        self._long_action_thread_lock = Lock()
+        self._async_executor = AsyncExecutor()
 
     def load_cluster_from_dcs(self):
-        cluster = self.dcs.get_cluster()
-
         # We want to keep the state of cluster when it was healhy
-        if cluster.is_unlocked() and self.cluster and not self.cluster.is_unlocked():
-            self.old_cluster = self.cluster
-        if not self.old_cluster:
-            self.old_cluster = cluster
-        self.cluster = cluster
+        if not self.dcs.get_cluster().is_unlocked() or not self.old_cluster:
+            self.old_cluster = self.dcs.cluster
 
     def acquire_lock(self):
         return self.dcs.attempt_to_acquire_leader()
@@ -44,26 +37,26 @@ class Ha:
         return ret
 
     def has_lock(self):
-        lock_owner = self.cluster.leader and self.cluster.leader.name
+        lock_owner = self.dcs.cluster.leader and self.dcs.cluster.leader.name
         logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name)
         return lock_owner == self.state_handler.name
 
-    def _run_async(self, func, args=()):
-        try:
-            return func(*args) if args else func()
-        except:
-            logger.exception('Exception during execution of long running task %s', self.get_scheduled_action())
-        finally:
-            with self._long_action_thread_lock:
-                self._long_action_in_progress = False
-                self._reset_scheduled_action()
-
-    def run_async(self, func, args=()):
-        self._long_action_in_progress = True
-        Thread(target=self._run_async, args=(func, args)).start()
+    def touch_member(self):
+        data = {
+            'conn_url': self.state_handler.connection_string,
+            'api_url': self.patroni.api.connection_string,
+            'state': self.state_handler.state,
+            'role': self.state_handler.role
+        }
+        if data['state'] in ['running', 'restarting', 'starting']:
+            try:
+                data['xlog_location'] = self.state_handler.xlog_position()
+            except:
+                pass
+        self.dcs.touch_member(json.dumps(data, separators=(',', ':')))
 
     def copy_backup_from_leader(self):
-        if self.state_handler.bootstrap(self.cluster.leader):
+        if self.state_handler.bootstrap(self.dcs.cluster.leader):
             logger.info('bootstrapped from leader')
         else:
             self.state_handler.stop('immediate')
@@ -71,15 +64,14 @@ class Ha:
             logger.error('failed to bootstrap from leader')
 
     def bootstrap(self):
-        if not self.cluster.is_unlocked():  # cluster already has leader
-            if self._long_action_in_progress:
+        if not self.dcs.cluster.is_unlocked():  # cluster already has leader
+            if self._async_executor.busy:
                 self.copy_backup_from_leader()
             else:
-                with self._scheduled_action_lock:
-                    self._scheduled_action = 'bootstrap from leader'
-                self.run_async(self.copy_backup_from_leader)
+                self._async_executor.schedule('bootstrap from leader')
+                self._async_executor.run_async(self.copy_backup_from_leader)
             return 'trying to bootstrap from leader'
-        elif not self.cluster.initialize:  # no initialize key
+        elif not self.dcs.cluster.initialize:  # no initialize key
             if self.dcs.initialize():  # race for initialization
                 try:
                     self.state_handler.bootstrap()
@@ -99,7 +91,7 @@ class Ha:
 
     def recover(self):
         has_lock = self.has_lock()
-        self.state_handler.write_recovery_conf(None if has_lock else self.cluster.leader)
+        self.state_handler.write_recovery_conf(None if has_lock else self.dcs.cluster.leader)
         if not self.state_handler.start():
             if not has_lock:
                 return 'failed to start postgres'
@@ -113,10 +105,9 @@ class Ha:
     def follow_the_leader(self, demote_reason, follow_reason, refresh=True):
         refresh and self.load_cluster_from_dcs()
         ret = demote_reason if self.state_handler.is_leader() else follow_reason
-        if not self.state_handler.check_recovery_conf(self.cluster.leader):
-            with self._scheduled_action_lock:
-                self._scheduled_action = 'changing primary_conninfo and restarting'
-            self.run_async(self.state_handler.follow_the_leader, (self.cluster.leader, ))
+        if not self.state_handler.check_recovery_conf(self.dcs.cluster.leader):
+            self._async_executor.schedule('changing primary_conninfo and restarting')
+            self._async_executor.run_async(self.state_handler.follow_the_leader, (self.dcs.cluster.leader, ))
         return ret
 
     def enforce_master_role(self, message, promote_message):
@@ -159,7 +150,7 @@ class Ha:
         if self.state_handler.is_leader():
             return True
 
-        if check_replication_lag and not self.state_handler.check_replication_lag(self.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
@@ -190,13 +181,13 @@ class Ha:
         return ret
 
     def manual_failover_process_no_leader(self):
-        failover = self.cluster.failover
+        failover = self.dcs.cluster.failover
         if failover.member:  # manual failover to specific member
             if failover.member == self.state_handler.name:  # manual failover to me
                 return True
 
             # find specific node and check that it is healthy
-            members = [m for m in self.cluster.members if m.name == failover.member]
+            members = [m for m in self.dcs.cluster.members if m.name == failover.member]
             if members:
                 member, reachable, in_recovery, xlog_location = self.fetch_node_status(members[0])
                 if reachable:  # node is healthy
@@ -212,7 +203,7 @@ class Ha:
         if failover.leader:
             if self.state_handler.name == failover.leader:  # I was the leader
                 # exclude me and desired member which is unhealthy (failover.member can be None)
-                members = [m for m in self.cluster.members if m.name != failover.member]
+                members = [m for m in self.dcs.cluster.members if m.name != failover.member]
                 if self.is_failover_possible(members):  # check that there are healthy members
                     return False
                 else:  # I was the leader and it looks like currently I am the only healthy member
@@ -221,15 +212,15 @@ class Ha:
             # at this point we assume that our node is a candidate for a failover among all nodes except former leader
 
         # exclude former leader from the list (failover.leader can be None)
-        members = [m for m in self.cluster.members if m.name != failover.leader]
+        members = [m for m in self.dcs.cluster.members if m.name != failover.leader]
         return self._is_healthiest_node(members, check_replication_lag=False)
 
     def is_healthiest_node(self):
-        if self.cluster.failover:
+        if self.dcs.cluster.failover:
             return self.manual_failover_process_no_leader()
 
         # run usual health check
-        members = {m.name: m for m in self.old_cluster.members + self.cluster.members}
+        members = {m.name: m for m in self.dcs.cluster.members + self.old_cluster.members}
         return self._is_healthiest_node(members.values())
 
     def demote(self, delete_leader=True):
@@ -239,14 +230,13 @@ class Ha:
         self.state_handler.follow_the_leader(None)
 
     def process_manual_failover_from_leader(self):
-        failover = self.cluster.failover
+        failover = self.dcs.cluster.failover
         if not failover.leader or failover.leader == self.state_handler.name:
             if not failover.member or failover.member != self.state_handler.name:
-                members = [m for m in self.cluster.members if not failover.member or m.name == failover.member]
+                members = [m for m in self.dcs.cluster.members if not failover.member or m.name == failover.member]
                 if self.is_failover_possible(members):  # check that there are healthy members
-                    with self._scheduled_action_lock:
-                        self._scheduled_action = 'manual failover: demote'
-                    self.run_async(self.demote)
+                    self._async_executor.schedule('manual failover: demote')
+                    self._async_executor.run_async(self.demote)
                     return 'manual failover: demoting myself'
                 else:
                     logger.warning('manual failover: no healthy members found, failover is not possible')
@@ -254,15 +244,15 @@ class Ha:
                 logger.warning('manual failover: I am already the leader, no need to failover')
         else:
             logger.warning('manual failover: leader name does not match: %s != %s',
-                           self.cluster.failover.leader, self.state_handler.name)
+                           self.dcs.cluster.failover.leader, self.state_handler.name)
 
         logger.info('Trying to clean up failover key')
-        self.dcs.manual_failover('', '', self.cluster.failover.index)
+        self.dcs.manual_failover('', '', self.dcs.cluster.failover.index)
 
     def process_unhealthy_cluster(self):
         if self.is_healthiest_node():
             if self.acquire_lock():
-                if self.cluster.failover:
+                if self.dcs.cluster.failover:
                     logger.info('Cleanning up failover key after acquiring leader lock...')
                     self.dcs.manual_failover('', '')
                 return self.enforce_master_role('acquired session lock as a leader',
@@ -276,7 +266,7 @@ class Ha:
 
     def process_healthy_cluster(self):
         if self.has_lock():
-            if self.cluster.failover:
+            if self.dcs.cluster.failover:
                 msg = self.process_manual_failover_from_leader()
                 if msg is not None:
                     return msg
@@ -293,42 +283,28 @@ class Ha:
         return self.follow_the_leader('demoting self because i do not have the lock and i was a leader',
                                       'no action.  i am a secondary and i am following a leader', False)
 
-    def schedule_action(self, action):
-        with self._long_action_thread_lock:
-            with self._scheduled_action_lock:
-                if self._scheduled_action is not None:
-                    return self._scheduled_action
-                self._scheduled_action = action
-        return None
-
-    def get_scheduled_action(self):
-        with self._scheduled_action_lock:
-            return self._scheduled_action
-
-    def _reset_scheduled_action(self):
-        with self._scheduled_action_lock:
-            self._scheduled_action = None
+    def schedule(self, action):
+        with self._async_executor:
+            return self._async_executor.schedule(action)
 
     def restart_scheduled(self):
-        return self.get_scheduled_action() == 'restart'
+        return self._async_executor.scheduled_action == 'restart'
 
     def schedule_reinitialize(self):
-        return self.schedule_action('reinitialize')
+        return self.schedule('reinitialize')
 
     def reinitialize_scheduled(self):
-        return self.get_scheduled_action() == 'reinitialize'
+        return self._async_executor.scheduled_action == 'reinitialize'
 
     def restart(self):
-        with self._long_action_thread_lock:
-            with self._scheduled_action_lock:
-                if self._scheduled_action is not None:
-                    return False, self._scheduled_action + ' already in progress'
-                self._scheduled_action = 'restart'
-            self._long_action_in_progress = True
-        if self._run_async(self.state_handler.restart):
-            return True, 'restarted successfully'
+        with self._async_executor:
+            prev = self._async_executor.schedule('restart', True)
+            if prev is not None:
+                return (False, prev + ' already in progress')
+        if self._async_executor.run(self.state_handler.restart):
+            return (True, 'restarted successfully')
         else:
-            return False, 'restart failed'
+            return (False, 'restart failed')
 
     def reinitialize(self):
         self.state_handler.stop('immediate')
@@ -338,47 +314,50 @@ class Ha:
 
     def process_scheduled_action(self):
         if self.reinitialize_scheduled():
-            if self.cluster.is_unlocked():
+            if self.dcs.cluster.is_unlocked():
                 logger.error('Cluster has no leader, can not reinitialize')
-                self._reset_scheduled_action()
+                self._async_executor.reset_scheduled_action()
             elif self.has_lock():
                 logger.error('I am the leader, can not reinitialize')
-                self._reset_scheduled_action()
+                self._async_executor.reset_scheduled_action()
             else:
-                self.run_async(self.reinitialize)
-                return True
+                self._async_executor.run_async(self.reinitialize)
+                return 'reinitialize started'
 
     def handle_long_action_in_progress(self):
         if self.has_lock():
             if self.update_lock():
-                return 'updated leader lock during ' + self.get_scheduled_action()
+                return 'updated leader lock during ' + self._async_executor.scheduled_action
             else:
-                return 'failed to update leader lock during ' + self.get_scheduled_action()
-        elif self.cluster.is_unlocked():
+                return 'failed to update leader lock during ' + self._async_executor.scheduled_action
+        elif self.dcs.cluster.is_unlocked():
             return 'not healthy enough for leader race'
         else:
-            return self.get_scheduled_action() + ' in progress'
+            return self._async_executor.scheduled_action + ' in progress'
 
     def _run_cycle(self):
         try:
             self.load_cluster_from_dcs()
 
+            self.touch_member()
+
             # cluster has leader key but not initialize key
-            if not self.cluster.is_unlocked() and not self.cluster.initialize:
+            if not self.dcs.cluster.is_unlocked() and not self.dcs.cluster.initialize:
                 self.dcs.initialize()  # fix it
 
-            if self._long_action_in_progress:
+            if self._async_executor.busy:
                 return self.handle_long_action_in_progress()
 
             # currently it can trigger only reinitialize
-            if self.process_scheduled_action():
-                return 'reinitialize started'
+            msg = self.process_scheduled_action()
+            if msg is not None:
+                return msg
 
             # is data directory empty?
             if self.state_handler.data_directory_empty():
                 return self.bootstrap()  # new node
             # "bootstrap", but data directory is not empty
-            elif not self.cluster.initialize and self.cluster.is_unlocked():
+            elif not self.dcs.cluster.initialize and self.dcs.cluster.is_unlocked():
                 self.dcs.initialize()
 
             # try to start dead postgres
@@ -388,12 +367,12 @@ class Ha:
                     return msg
 
             try:
-                if self.cluster.is_unlocked():
+                if self.dcs.cluster.is_unlocked():
                     return self.process_unhealthy_cluster()
                 else:
                     return self.process_healthy_cluster()
             finally:
-                self.state_handler.sync_replication_slots(self.cluster)
+                self.state_handler.sync_replication_slots(self.dcs.cluster)
         except DCSError:
             logger.error('Error communicating with DCS')
             if self.state_handler.is_running() and self.state_handler.is_leader():
@@ -403,5 +382,5 @@ class Ha:
             logger.exception('Error communicating with Postgresql. Will try again later')
 
     def run_cycle(self):
-        with self._long_action_thread_lock:
+        with self._async_executor:
             return self._run_cycle()

patroni/postgresql.py

@@ -68,7 +68,7 @@ class Postgresql:
         self._connection = None
         self._cursor_holder = None
         self.replication_slots = []  # list of already existing replication slots
-        self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException)
+        self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException)
 
         self._state = 'stopped'
         self._state_lock = Lock()
@@ -381,7 +381,7 @@ recovery_target_timeline = 'latest'
         return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
                                                                THEN pg_last_xlog_replay_location()
                                                                ELSE pg_current_xlog_location()
-                                                          END, '0/0')""").fetchone()[0]
+                                                          END, '0/0')::bigint""").fetchone()[0]
 
     def load_replication_slots(self):
         if self.use_slots and self.schedule_load_slots:

patroni/utils.py

@@ -36,6 +36,8 @@ def calculate_ttl(expiration):
     """
     >>> calculate_ttl(None)
     >>> calculate_ttl('2015-06-10 12:56:30.552539016Z')
+    >>> calculate_ttl('2015-06-10T12:56:30.552539016Z') < 0
+    True
     """
     if not expiration:
         return None

patroni/zookeeper.py

@@ -5,7 +5,7 @@ import time
 
 from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
-from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member, parse_connection_string
+from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member
 from patroni.exceptions import DCSError
 from patroni.utils import sleep
 from requests.exceptions import RequestException
@@ -93,7 +93,7 @@ class ZooKeeper(AbstractDCS):
         self.client.add_listener(self.session_listener)
         self.cluster_event = self.client.handler.event_object()
 
-        self.cluster = None
+        self._my_member_data = None
         self.fetch_cluster = True
         self.last_leader_operation = 0
 
@@ -116,8 +116,7 @@ class ZooKeeper(AbstractDCS):
 
     @staticmethod
     def member(name, value, znode):
-        conn_url, api_url = parse_connection_string(value)
-        return Member(znode.version, name, conn_url, api_url, None, None)
+        return Member.from_node(znode.version, name, znode.ephemeralOwner, value)
 
     def get_children(self, key, watch=None):
         try:
@@ -153,9 +152,9 @@ class ZooKeeper(AbstractDCS):
                 leader = None
 
             if leader:
-                member = Member(-1, leader[0], None, None, None, None)
+                member = Member(-1, leader[0], None, {})
                 member = ([m for m in members if m.name == leader[0]] or [member])[0]
-                leader = Leader(leader[1].version, None, None, member)
+                leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
                 self.fetch_cluster = member.index == -1
 
         # failover key
@@ -207,21 +206,34 @@ class ZooKeeper(AbstractDCS):
     def initialize(self):
         return self._create(self.initialize_path, self._name, makepath=True)
 
-    def touch_member(self, connection_string, ttl=None):
-        if not self.fetch_cluster and self.cluster and any(m.name == self._name for m in self.cluster.members):
-            return True
+    def touch_member(self, data, ttl=None):
+        me = self.cluster and ([m for m in self.cluster.members if m.name == self._name] or [None])[0]
         path = self.member_path
-        connection_string = connection_string.encode('utf-8')
+        data = data.encode('utf-8')
+        create = not me
+        if me and self.client.client_id is not None and me.session != self.client.client_id[0]:
+            try:
+                self.client.retry(self.client.delete, path)
+            except NoNodeError:
+                pass
+            except:
+                return False
+            create = True
+
+        if not create and data == self._my_member_data:
+            return True
+
         try:
-            self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True)
+            if create:
+                self.client.retry(self.client.create, path, data, makepath=True, ephemeral=True)
+            else:
+                self.client.retry(self.client.set, path, data)
+            self._my_member_data = data
             return True
         except NodeExistsError:
             try:
-                node = self.get_node(path)
-                if node and self.client.client_id is not None and node[1].ephemeralOwner == self.client.client_id[0]:
-                    return True
-                self.client.retry(self.client.delete, path)
-                self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True)
+                self.client.retry(self.client.set, path, data)
+                self._my_member_data = data
                 return True
             except:
                 logger.exception('touch_member')
@@ -252,6 +264,7 @@ class ZooKeeper(AbstractDCS):
 
     def delete_leader(self):
         self.client.restart()
+        self._my_member_data = None
         return True
 
     def _cancel_initialization(self):

tests/test_api.py

@@ -43,6 +43,7 @@ class MockPatroni:
 
     postgresql = MockPostgresql()
     ha = MockHa()
+    dcs = Mock()
 
 
 class MockRequest:

tests/test_etcd.py

@@ -149,15 +149,6 @@ def http_request(method, url, **kwargs):
     raise socket.error
 
 
-class TestMember(unittest.TestCase):
-
-    def test_real_ttl(self):
-        now = datetime.datetime.utcnow()
-        member = Member(0, 'a', 'b', 'c', (now + datetime.timedelta(seconds=2)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None)
-        self.assertLess(member.real_ttl(), 2)
-        self.assertEquals(Member(0, 'a', 'b', 'c', '', None).real_ttl(), -1)
-
-
 @patch('dns.resolver.query', dns_query)
 @patch('socket.getaddrinfo', socket_getaddrinfo)
 @patch('requests.get', requests_get)

tests/test_ha.py

@@ -26,11 +26,11 @@ def get_cluster_not_initialized_without_leader():
 
 
 def get_cluster_initialized_without_leader(leader=False, failover=None):
-    m = Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
-               'http://127.0.0.1:8008/patroni', None, 28)
-    l = Leader(0, 0, 0, m) if leader else None
-    o = Member(0, 'other', 'postgres://replicator:[email protected]:5436/postgres',
-               'http://127.0.0.1:8011/patroni', None, 28)
+    m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres',
+                                 'api_url': 'http://127.0.0.1:8008/patroni'})
+    l = Leader(0, 0, m) if leader else None
+    o = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+                                'api_url': 'http://127.0.0.1:8011/patroni'})
     return get_cluster(True, l, [m, o], failover)
 
 
@@ -42,6 +42,8 @@ class MockPostgresql(Mock):
 
     name = 'postgresql0'
     role = 'replica'
+    state = 'running'
+    connection_string = 'postgres://foo@bar/postgres'
 
     def is_healthy(self):
         return True
@@ -74,6 +76,14 @@ class MockPostgresql(Mock):
         return False
 
 
+class MockPatroni:
+
+    def __init__(self, p, d):
+        self.postgresql = p
+        self.dcs = d
+        self.api = Mock()
+        self.api.connection_string = 'http://127.0.0.1:8008'
+
 def run_async(func, args=()):
     func(args) if args else func()
 
@@ -88,22 +98,20 @@ class TestHa(unittest.TestCase):
         self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'})
         self.e.client.read = etcd_read
         self.e.client.write = etcd_write
-        self.ha = Ha(self.p, self.e)
-        self.ha.run_async = run_async
-        self.ha.load_cluster_from_dcs()
-        self.ha.cluster = get_cluster_not_initialized_without_leader()
+        self.ha = Ha(MockPatroni(self.p, self.e))
+        self.ha._async_executor.run_async = run_async
+        self.ha.old_cluster = self.e.get_cluster()
+        self.e.cluster = get_cluster_not_initialized_without_leader()
         self.ha.load_cluster_from_dcs = Mock()
 
-    def test_load_cluster_from_dcs(self):
-        ha = Ha(self.p, self.e)
-        ha.load_cluster_from_dcs()
-        self.e.get_cluster = get_cluster_not_initialized_without_leader
-        ha.load_cluster_from_dcs()
-
     def test_update_lock(self):
         self.p.last_operation = Mock(side_effect=PostgresException(''))
         self.assertTrue(self.ha.update_lock())
 
+    def test_touch_member(self):
+        self.p.xlog_position = Mock(side_effect=Exception)
+        self.ha.touch_member()
+
     def test_start_as_replica(self):
         self.p.is_healthy = false
         self.assertEquals(self.ha.run_cycle(), 'started as a secondary')
@@ -119,8 +127,8 @@ class TestHa(unittest.TestCase):
         self.ha.has_lock = true
         self.assertEquals(self.ha.run_cycle(), 'removed leader key after trying and failing to start postgres')
 
+    @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_start_as_readonly(self):
-        self.ha.cluster.is_unlocked = false
         self.p.is_leader = self.p.is_healthy = false
         self.ha.has_lock = true
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock')
@@ -153,28 +161,28 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
 
     def test_promote_because_have_lock(self):
-        self.ha.cluster.is_unlocked = false
+        self.e.cluster.is_unlocked = false
         self.ha.has_lock = true
         self.p.is_leader = false
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock')
 
     def test_leader_with_lock(self):
-        self.ha.cluster.is_unlocked = false
+        self.e.cluster.is_unlocked = false
         self.ha.has_lock = true
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
 
     def test_demote_because_not_having_lock(self):
-        self.ha.cluster.is_unlocked = false
+        self.e.cluster.is_unlocked = false
         self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader')
 
     def test_demote_because_update_lock_failed(self):
-        self.ha.cluster.is_unlocked = false
+        self.e.cluster.is_unlocked = false
         self.ha.has_lock = true
         self.ha.update_lock = false
         self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader')
 
     def test_follow_the_leader(self):
-        self.ha.cluster.is_unlocked = false
+        self.e.cluster.is_unlocked = false
         self.p.is_leader = false
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
 
@@ -182,36 +190,28 @@ class TestHa(unittest.TestCase):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
         self.assertEquals(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader')
 
-    def test__run_async(self):
-        self.ha._run_async(Mock(side_effect=Exception()))        
-
-    @patch.object(Thread, 'start', Mock())
-    def test_run_async(self):
-        ha = Ha(self.p, self.e)
-        ha.run_async(true)
-
     def test_bootstrap_from_leader(self):
-        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.e.cluster = get_cluster_initialized_with_leader()
         self.p.bootstrap = false
         self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from leader')
-        self.ha._long_action_in_progress = True
+        self.ha._async_executor._busy = True
         self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from leader')
 
     def test_bootstrap_waiting_for_leader(self):
-        self.ha.cluster = get_cluster_initialized_without_leader()
+        self.e.cluster = get_cluster_initialized_without_leader()
         self.assertEquals(self.ha.bootstrap(), 'waiting for leader to bootstrap')
 
     def test_bootstrap_initialize_lock_failed(self):
-        self.ha.cluster = get_cluster_not_initialized_without_leader()
+        self.e.cluster = get_cluster_not_initialized_without_leader()
         self.assertEquals(self.ha.bootstrap(), 'failed to acquire initialize lock')
 
     def test_bootstrap_initialized_new_cluster(self):
-        self.ha.cluster = get_cluster_not_initialized_without_leader()
+        self.e.cluster = get_cluster_not_initialized_without_leader()
         self.e.initialize = true
         self.assertEquals(self.ha.bootstrap(), 'initialized a new cluster')
 
     def test_bootstrap_release_initialize_key_on_failure(self):
-        self.ha.cluster = get_cluster_not_initialized_without_leader()
+        self.e.cluster = get_cluster_not_initialized_without_leader()
         self.e.initialize = true
         self.p.bootstrap = Mock(side_effect=PostgresException("Could not bootstrap master PostgreSQL"))
         self.assertRaises(PostgresException, self.ha.bootstrap)
@@ -220,13 +220,13 @@ class TestHa(unittest.TestCase):
         self.ha.schedule_reinitialize()
         self.ha.schedule_reinitialize()
         self.ha.run_cycle()
-        self.assertIsNone(self.ha.get_scheduled_action())
+        self.assertIsNone(self.ha._async_executor.scheduled_action)
 
-        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.e.cluster = get_cluster_initialized_with_leader()
         self.ha.has_lock = true
         self.ha.schedule_reinitialize()
         self.ha.run_cycle()
-        self.assertIsNone(self.ha.get_scheduled_action())
+        self.assertIsNone(self.ha._async_executor.scheduled_action)
 
         self.ha.has_lock = false
         self.ha.schedule_reinitialize()
@@ -240,12 +240,11 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.restart(), (False, 'reinitialize already in progress'))
 
     def test_restart_in_progress(self):
-        self.ha._long_action_in_progress = True
-        self.ha._scheduled_action = 'restart'
+        self.ha._async_executor.schedule('restart', True)
         self.assertTrue(self.ha.restart_scheduled())
         self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race')
 
-        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.e.cluster = get_cluster_initialized_with_leader()
         self.assertEquals(self.ha.run_cycle(), 'restart in progress')
 
         self.ha.has_lock = true
@@ -257,26 +256,26 @@ class TestHa(unittest.TestCase):
     @patch('requests.get', requests_get)
     def test_manual_failover_from_leader(self):
         self.ha.has_lock = true
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', ''))
+        self.e.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', ''))
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', MockPostgresql.name))
+        self.e.cluster = get_cluster_initialized_with_leader(Failover(0, '', MockPostgresql.name))
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla'))
+        self.e.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla'))
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
         f = Failover(0, MockPostgresql.name, '')
-        self.ha.cluster = get_cluster_initialized_with_leader(f)
+        self.e.cluster = get_cluster_initialized_with_leader(f)
         self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself')
 
     @patch('requests.get', requests_get)
     def test_manual_failover_process_no_leader(self):
         self.p.is_leader = false
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', MockPostgresql.name))
+        self.e.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', MockPostgresql.name))
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader'))
+        self.e.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader'))
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
         self.ha.fetch_node_status = lambda e: (e, True, True, 0)  # accessible, in_recovery
         self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, MockPostgresql.name, ''))
+        self.e.cluster = get_cluster_initialized_without_leader(failover=Failover(0, MockPostgresql.name, ''))
         self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
         self.ha.fetch_node_status = lambda e: (e, False, True, 0)  # accessible, in_recovery
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
@@ -295,7 +294,7 @@ class TestHa(unittest.TestCase):
 
     @patch('requests.get', requests_get)
     def test_fetch_node_status(self):
-        member = Member(0, 'test', '', 'http://127.0.0.1:8011/patroni', None, None)
+        member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'})
         self.ha.fetch_node_status(member)
-        member = Member(0, 'test', '', 'http://localhost:8011/patroni', None, None)
+        member = Member(0, 'test', 1, {'api_url': 'http://localhost:8011/patroni'})
         self.ha.fetch_node_status(member)

tests/test_patroni.py

@@ -6,6 +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
@@ -27,7 +28,7 @@ def time_sleep(*args):
 @patch.object(Postgresql, 'write_pg_hba', Mock())
 @patch.object(Postgresql, 'write_recovery_conf', Mock())
 @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
-@patch.object(Ha, 'run_async', Mock())
+@patch.object(AsyncExecutor, 'run', Mock())
 class TestPatroni(unittest.TestCase):
 
     @patch.object(Client, 'machines')
@@ -57,15 +58,13 @@ class TestPatroni(unittest.TestCase):
         sys.argv = ['patroni.py', 'postgres0.yml']
 
         mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
-        with patch.object(Patroni, 'touch_member', self.touch_member):
-            with patch.object(Patroni, 'run', Mock(side_effect=SleepException())):
-                self.assertRaises(SleepException, main)
-            with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
-                main()
+        with patch.object(Patroni, 'run', Mock(side_effect=SleepException())):
+            self.assertRaises(SleepException, main)
+        with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
+            main()
 
     @patch('time.sleep', Mock(side_effect=SleepException()))
     def test_run(self):
-        self.p.touch_member = self.touch_member
         self.p.ha.dcs.watch = time_sleep
         self.assertRaises(SleepException, self.p.run)
 
@@ -73,20 +72,6 @@ class TestPatroni(unittest.TestCase):
         self.p.api.start = Mock()
         self.assertRaises(SleepException, self.p.run)
 
-    def touch_member(self, ttl=None):
-        if not self.touched:
-            self.touched = True
-            return False
-        return True
-
-    def test_touch_member(self):
-        self.p.touch_member()
-        now = datetime.datetime.utcnow()
-        member = Member(0, self.p.postgresql.name, 'b', 'c', (now + datetime.timedelta(
-            seconds=self.p.shutdown_member_ttl + 10)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None)
-        self.p.ha.cluster = Cluster(True, member, 0, [member], None)
-        self.p.touch_member()
-
     def test_schedule_next_run(self):
         self.p.ha.dcs.watch = Mock(return_value=True)
         self.p.schedule_next_run()

tests/test_postgresql.py

@@ -105,10 +105,10 @@ class TestPostgresql(unittest.TestCase):
                              'restore': 'true'})
         if not os.path.exists(self.p.data_dir):
             os.makedirs(self.p.data_dir)
-        self.leadermem = Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres', None, None, 28)
-        self.leader = Leader(-1, None, 28, self.leadermem)
-        self.other = Member(0, 'test1', 'postgres://replicator:[email protected]:5433/postgres', None, None, 28)
-        self.me = Member(0, 'test0', 'postgres://replicator:[email protected]:5434/postgres', None, None, 28)
+        self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
+        self.leader = Leader(-1, 28, self.leadermem)
+        self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres'})
+        self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:[email protected]:5434/postgres'})
 
     def tearDown(self):
         shutil.rmtree('data')
@@ -146,7 +146,7 @@ class TestPostgresql(unittest.TestCase):
         self.p.follow_the_leader(self.leader)
         self.p.demote()
         self.p.follow_the_leader(self.leader)
-        self.p.follow_the_leader(Leader(-1, None, 28, self.other))
+        self.p.follow_the_leader(Leader(-1, 28, self.other))
 
     def test_create_replica(self):
         self.p.delete_trigger_file = Mock(side_effect=OSError())

tests/test_zookeeper.py

@@ -69,6 +69,9 @@ class MockKazooClient(Mock):
             raise TypeError("Invalid type for 'value' (must be a byte string)")
         if path == '/service/bla/optime/leader':
             raise Exception
+        if path == '/service/test/members/bar':
+            if value == b'retry':
+                return
         if path == '/service/test/failover':
             if value == b'Exception':
                 raise Exception
@@ -85,7 +88,9 @@ class MockKazooClient(Mock):
                 return
             self.leader = True
             raise Exception
-        elif path.endswith('/initialize'):
+        elif path == '/service/test/members/buzz':
+            raise Exception
+        elif path.endswith('/initialize') or path == '/service/test/members/bar':
             raise NoNodeError
 
 
@@ -137,10 +142,18 @@ class TestZooKeeper(unittest.TestCase):
         self.zk.cancel_initialization()
 
     def test_touch_member(self):
+        self.zk._name = 'buzz'
+        self.zk.get_cluster()
         self.zk.touch_member('new')
+        self.zk._name = 'bar'
+        self.zk.touch_member('new')
+        self.zk._name = 'na'
+        self.zk.client.exists = 1
         self.zk.touch_member('exists')
+        self.zk._name = 'bar'
         self.zk.touch_member('retry')
-        self.zk.client.exists = True
+        self.zk.fetch_cluster = True
+        self.zk.get_cluster()
         self.zk.touch_member('retry')
 
     def test_take_leader(self):