Merge branch 'master' of github.com:zalando/patroni into restore/movebasebackup

Conflicts:
	patroni/postgresql.py

.gitignore

@@ -1,3 +1,11 @@
 data/*
 *.pyc
-helpers/*.pyc
+*.egg/
+*.egg-info/
+.cache/
+.coverage
+.eggs/
+build/
+coverage.xml
+junit.xml
+pgpass

docker/README.md

@@ -1,7 +1,7 @@
 # Patroni Dockerfile
 You can run Patroni in a docker container using this Dockerfile, or by using one of the Docker image at
 
-    https://os-registry.stups.zalan.do/v1/repositories/acid/patroni/tags
+    https://registry.opensource.zalan.do/v1/repositories/acid/patroni/tags
 
 This Dockerfile is meant in aiding development of Patroni and quick testing of features. It is not a production-worthy
 Dockerfile
@@ -10,7 +10,7 @@ Dockerfile
 
 ## Standalone Patroni
 
-    docker run -d os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT
+    docker run -d registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT
 
 ## Multiple Patroni's communicating with a standalone etcd inside Docker
 
@@ -36,12 +36,12 @@ To automate this you can run the following script:
 
 Example session:
 
-    $ ./dev_patroni_cluster.sh --image os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT --members=2 --name=bravo
+    $ ./dev_patroni_cluster.sh --image registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT --members=2 --name=bravo
     The etcd container is 6be871a11cb373406ca5ea1c6b39e1.0-SNAPSHOTfdde9fb1d6177212d6ad0c0d1bd9b563, ip=172.17.1.24
     Started Patroni container 67e611f2eca7c40f9e6e0e24a4a8f2cba7e3e56d22a420e15ab9240a37a9d7a4, ip=172.17.1.25
     Started Patroni container 47dd12ae635ab83b039f5889e250048b606ed5e48e3650b69e365e7e1d4acbcf, ip=172.17.1.26
     $ docker ps
     CONTAINER ID        IMAGE                                         COMMAND                CREATED             STATUS              PORTS                          NAMES
-    47dd12ae635a        os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT   "/bin/bash /entrypoi   10 seconds ago      Up 8 seconds        4001/tcp, 5432/tcp, 2380/tcp   bravo_OR64g8bx
+    47dd12ae635a        registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT   "/bin/bash /entrypoi   10 seconds ago      Up 8 seconds        4001/tcp, 5432/tcp, 2380/tcp   bravo_OR64g8bx
-    67e611f2eca7        os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT   "/bin/bash /entrypoi   11 seconds ago      Up 10 seconds       2380/tcp, 4001/tcp, 5432/tcp   bravo_si9no8iz
+    67e611f2eca7        registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT   "/bin/bash /entrypoi   11 seconds ago      Up 10 seconds       2380/tcp, 4001/tcp, 5432/tcp   bravo_si9no8iz
-    6be871a11cb3        os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT   "/bin/bash /entrypoi   12 seconds ago      Up 10 seconds       4001/tcp, 5432/tcp, 2380/tcp   bravo_etcd
+    6be871a11cb3        registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT   "/bin/bash /entrypoi   12 seconds ago      Up 10 seconds       4001/tcp, 5432/tcp, 2380/tcp   bravo_etcd

docker/dev_patroni_cluster.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-DOCKER_IMAGE="os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT"
+DOCKER_IMAGE="registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT"
 MEMBERS=3
 
 

docker/entrypoint.sh

@@ -21,7 +21,7 @@ __EOF__
 }
 
 DOCKER_IP=$(hostname --ip-address)
-PATRONI_SCOPE=batman
+PATRONI_SCOPE=${PATRONI_SCOPE:-batman}
 
 optspec=":vh-:"
 while getopts "$optspec" optchar; do

patroni/__init__.py

@@ -8,7 +8,7 @@ from patroni.api import RestApiServer
 from patroni.etcd import Etcd
 from patroni.ha import Ha
 from patroni.postgresql import Postgresql
-from patroni.utils import setup_signal_handlers, sleep, reap_children
+from patroni.utils import setup_signal_handlers, reap_children
 from patroni.zookeeper import ZooKeeper
 
 logger = logging.getLogger(__name__)
@@ -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,30 +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 initialize(self):
-        # wait for etcd to be available
-        while not self.touch_member():
-            logger.info('waiting on DCS')
-            sleep(5)
-
-        self.postgresql.schedule_load_slots = self.postgresql.is_running() and self.postgresql.use_slots
-
     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):
@@ -64,12 +47,7 @@ class Patroni:
         self.next_run = time.time()
 
         while True:
-            self.touch_member()
             logger.info(self.ha.run_cycle())
-            try:
-                self.ha.cluster and self.ha.state_handler.sync_replication_slots(self.ha.cluster)
-            except:
-                logger.exception('Exception when changing replication slots')
             reap_children()
             self.schedule_next_run()
 
@@ -87,13 +65,11 @@ def main():
         config = yaml.load(f)
 
     patroni = Patroni(config)
-    patroni.initialize()
     try:
         patroni.run()
     except KeyboardInterrupt:
         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

@@ -3,6 +3,7 @@ import fcntl
 import json
 import logging
 import psycopg2
+import time
 
 from patroni.exceptions import PostgresConnectionException
 from patroni.utils import Retry, RetryFailedError
@@ -44,23 +45,35 @@ class RestApiHandler(BaseHTTPRequestHandler):
     def do_GET(self):
         """Default method for processing all GET requests which can not be routed to other methods"""
 
+        path = '/master' if self.path == '/' else self.path
         response = self.get_postgresql_status()
 
-        path = '/master' if self.path == '/' else self.path
+        patroni = self.server.patroni
-        status_code = 200 if response['running'] and 'role' in response and response['role'] in path else 503
+        cluster = patroni.dcs.cluster
+        if cluster:  # dcs available
+            if cluster.leader and 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
+            status_code = 200
+        else:
+            status_code = 503
 
         self.send_response(status_code)
         self.send_header('Content-Type', 'application/json')
         self.end_headers()
         self.wfile.write(json.dumps(response).encode('utf-8'))
 
-    @check_auth
-    def do_GET_sampleauth(self):
-        self.send_response(200)
-        self.send_header('Content-Type', 'text/html')
-        self.end_headers()
-        self.wfile.write(b'Hello!')
-
     def do_GET_patroni(self):
         response = self.get_postgresql_status(True)
 
@@ -69,6 +82,96 @@ class RestApiHandler(BaseHTTPRequestHandler):
         self.end_headers()
         self.wfile.write(json.dumps(response).encode('utf-8'))
 
+    @check_auth
+    def do_POST_restart(self):
+        status_code = 503
+        data = b'restart failed'
+        try:
+            status, msg = self.server.patroni.ha.restart()
+            status_code = 200 if status else 503
+            data = msg.encode('utf-8')
+        except:
+            logger.exception('Exception during restart')
+
+        self.send_response(status_code)
+        self.send_header('Content-Type', 'text/html')
+        self.end_headers()
+        self.wfile.write(data)
+
+    @check_auth
+    def do_POST_reinitialize(self):
+        ha = self.server.patroni.ha
+        cluster = ha.dcs.get_cluster()
+        if cluster.is_unlocked():
+            status_code = 503
+            data = b'Cluster has no leader, can not reinitialize'
+        elif cluster.leader.name == ha.state_handler.name:
+            status_code = 503
+            data = b'I am the leader, can not reinitialize'
+        else:
+            action = ha.schedule_reinitialize()
+            if action is not None:
+                status_code = 503
+                data = (action + ' already in progress').encode('utf-8')
+            else:
+                status_code = 200
+                data = b'reinitialize scheduled'
+
+        self.send_response(status_code)
+        self.send_header('Content-Type', 'text/html')
+        self.end_headers()
+        self.wfile.write(data)
+
+    def poll_failover_result(self, leader, member):
+        for a in range(0, 15):
+            time.sleep(1)
+            try:
+                cluster = self.server.patroni.dcs.get_cluster()
+                if cluster.leader and cluster.leader.name != leader:
+                    return 200, ('Successfully failed over to ' + cluster.leader.name).encode('utf-8')
+                if not cluster.failover:
+                    return 503, b'Failover failed'
+            except:
+                pass
+        return 503, b'Failover status unknown'
+
+    def is_failover_possible(self, cluster, leader, member):
+        if leader and not cluster.leader or cluster.leader.name != leader:
+            return b'leader name does not match'
+        if member:
+            members = [m for m in cluster.members if m.name == member]
+            if not members:
+                return b'member does not exists'
+        else:
+            members = [m for m in cluster.members if m.name != cluster.leader.name and m.api_url]
+            if not members:
+                return b'failover is not possible: cluster does not have members except leader'
+        for member, reachable, in_recovery, xlog_location in self.server.patroni.ha.fetch_nodes_statuses(members):
+            if reachable:
+                return None
+        return b'failover is not possible: no good candidates have been found'
+
+    @check_auth
+    def do_POST_failover(self):
+        content_length = int(self.headers.get('content-length', 0))
+        request = json.loads(self.rfile.read(content_length).decode('utf-8'))
+        leader = request.get('leader', None)
+        member = request.get('member', None)
+        cluster = self.server.patroni.ha.dcs.get_cluster()
+        status_code = 503
+        data = self.is_failover_possible(cluster, leader, member)
+        if not data:
+            if not self.server.patroni.dcs.manual_failover(leader, member):
+                data = b'failed to write failover key into DCS'
+            else:
+                self.server.patroni.dcs.event.set()
+                status_code, data = self.poll_failover_result(cluster.leader and cluster.leader.name, member)
+
+        self.send_response(status_code)
+        self.send_header('Content-Type', 'text/html')
+        self.end_headers()
+        self.wfile.write(data)
+
     def parse_request(self):
         """Override parse_request method to enrich basic functionality of `BaseHTTPRequestHandler` class
 
@@ -90,7 +193,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
     def query(self, sql, *params, **kwargs):
         if not kwargs.get('retry', False):
             return self.server.query(sql, *params)
-        retry = Retry(delay=2, retry_exceptions=PostgresConnectionException)
+        retry = Retry(delay=1, retry_exceptions=PostgresConnectionException)
         return retry(self.server.query, sql, *params)
 
     def get_postgresql_status(self, retry=False):
@@ -98,15 +201,16 @@ class RestApiHandler(BaseHTTPRequestHandler):
             row = self.query("""SELECT to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),
                                        pg_is_in_recovery(),
                                        CASE WHEN pg_is_in_recovery()
-                                            THEN null
+                                            THEN 0
-                                            ELSE pg_current_xlog_location() END,
+                                            ELSE pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint
-                                       pg_last_xlog_receive_location(),
+                                       END,
-                                       pg_last_xlog_replay_location(),
+                                       pg_xlog_location_diff(pg_last_xlog_receive_location(), '0/0')::bigint,
+                                       pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint,
                                        pg_is_in_recovery() AND pg_is_xlog_replay_paused()""", retry=retry)[0]
             return {
-                'running': True,
+                'state': self.server.patroni.postgresql.state,
                 'postmaster_start_time': row[0],
-                'role': 'slave' if row[1] else 'master',
+                'role': 'replica' if row[1] else 'master',
                 'xlog': ({
                     'received_location': row[3],
                     'replayed_location': row[4],
@@ -115,8 +219,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 })
             }
         except (psycopg2.Error, RetryFailedError, PostgresConnectionException):
+            state = self.server.patroni.postgresql.state
+            if state in ['stopped', 'starting', 'stopping', 'restarting', 'running']:
                 logger.exception('get_postgresql_status')
-            return {'running': self.server.patroni.postgresql.is_running()}
+                state = 'unknown' if state == 'running' else state
+            return {'state': state}
 
 
 class RestApiServer(ThreadingMixIn, HTTPServer, Thread):

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,9 +1,10 @@
 import abc
+import json
 
 from collections import namedtuple
 from patroni.exceptions import DCSError
-from patroni.utils import calculate_ttl, sleep
 from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
+from threading import Event, Lock
 
 
 def parse_connection_string(value):
@@ -23,28 +24,52 @@ 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 session: either session id or just ttl in seconds
-    :param api_url: REST API url of patroni instance
+    :param data: arbitrary data i.e. conn_url, api_url, xlog location, state, role, tags, etc...
-    :param expiration: expiration time of given member key
-    :param ttl: ttl of given member key in seconds"""
 
-    def real_ttl(self):
+    There are two mandatory keys in a data:
-        return calculate_ttl(self.expiration) or -1
+    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
+        >>> 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:
+            try:
+                data = json.loads(data)
+            except:
+                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 session: either session id or just ttl in seconds
-    :param ttl: ttl of the leader key
     :param member: reference to a `Member` object which represents current leader (see `Cluster.members`)"""
 
     @property
@@ -56,7 +81,15 @@ class Leader(namedtuple('Leader', 'index,expiration,ttl,member')):
         return self.member.conn_url
 
 
-class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')):
+class Failover(namedtuple('Failover', 'index,leader,member')):
+
+    @staticmethod
+    def from_node(index, value):
+        t = [a.strip() for a in value.split(':')] + ['']
+        return Failover(index, t[0], t[1]) if t[0] or t[1] else None
+
+
+class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members,failover')):
 
     """Immutable object (namedtuple) which represents PostgreSQL cluster.
     Consists of the following fields:
@@ -64,7 +97,8 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
     :param leader: `Leader` object which represents current leader of the cluster
     :param last_leader_operation: int or long object containing position of last known leader operation.
         This value is stored in `/optime/leader` key
-    :param members: list of Member object, all PostgreSQL cluster members including leader"""
+    :param members: list of Member object, all PostgreSQL cluster members including leader
+    :param failover: reference to `Failover` object"""
 
     def is_unlocked(self):
         return not (self.leader and self.leader.name)
@@ -76,6 +110,7 @@ class AbstractDCS:
 
     _INITIALIZE = 'initialize'
     _LEADER = 'leader'
+    _FAILOVER = 'failover'
     _MEMBERS = 'members/'
     _OPTIME = 'optime'
     _LEADER_OPTIME = _OPTIME + '/' + _LEADER
@@ -90,6 +125,10 @@ class AbstractDCS:
         self._scope = config['scope']
         self._base_path = '/service/' + self._scope
 
+        self._cluster = None
+        self._cluster_thread_lock = Lock()
+        self.event = Event()
+
     def client_path(self, path):
         return '/'.join([self._base_path, path.lstrip('/')])
 
@@ -109,25 +148,54 @@ class AbstractDCS:
     def leader_path(self):
         return self.client_path(self._LEADER)
 
+    @property
+    def failover_path(self):
+        return self.client_path(self._FAILOVER)
+
     @property
     def leader_optime_path(self):
         return self.client_path(self._LEADER_OPTIME)
 
     @abc.abstractmethod
+    def _load_cluster(self):
+        """Internally this method should build  `Cluster` object which
+           represents current state and topology of the cluster in DCS.
+           this method supposed to be called only by `get_cluster` method.
+
+           raise `~DCSError` in case of communication or other problems with DCS.
+           If the current node was running as a master and exception raised,
+           instance would be demoted."""
+
     def get_cluster(self):
-        """:returns: `Cluster` object which represent current state and topology of the cluster
+        with self._cluster_thread_lock:
-        raise `~DCSError` in case of communication or other problems with DCS. If current instance was
+            try:
-            running as a master and exception raised instance would be demoted."""
+                self._load_cluster()
+            except:
+                self._cluster = None
+                raise
+            return self._cluster
+
+    @property
+    def cluster(self):
+        with self._cluster_thread_lock:
+            return self._cluster
+
+    def reset_cluster(self):
+        with self._cluster_thread_lock:
+            self._cluster = None
 
     @abc.abstractmethod
-    def update_leader(self, state_handler):
+    def write_leader_optime(self, last_operation):
-        """Update leader key (or session) ttl and `/optime/leader` key in DCS.
+        """write current xlog location into `/optime/leader` key in DCS
+        :param last_operation: absolute xlog location in bytes"""
+
+    @abc.abstractmethod
+    def update_leader(self):
+        """Update leader key (or session) ttl
 
-        :param state_handler: reference to `Postgresql` object
         :returns: `!True` if leader key (or session) has been updated successfully.
             If not, `!False` must be returned and current instance would be demoted.
 
-        If you failed to update `/optime/leader` this error is not critical and you can return `!True`
         You have to use CAS (Compare And Swap) operation in order to update leader key,
         for example for etcd `prevValue` parameter must be used."""
 
@@ -140,6 +208,13 @@ class AbstractDCS:
         Key must be created atomically. In case if key already exists it should not be
         overwritten and `!False` must be returned"""
 
+    @abc.abstractmethod
+    def set_failover_value(self, value, index=None):
+        """Create or update `/failover` key"""
+
+    def manual_failover(self, leader, member, index=None):
+        return self.set_failover_value(leader + (':' + member if member else ''), index)
+
     def current_leader(self):
         try:
             cluster = self.get_cluster()
@@ -165,8 +240,11 @@ class AbstractDCS:
         overwriting the key if necessary."""
 
     @abc.abstractmethod
-    def initialize(self):
+    def initialize(self, create_new=True, sysid=""):
         """Race for cluster initialization.
+
+        :param create_new: False if the key should already exist (in the case we are setting the system_id)
+        :param sysid: PostgreSQL cluster system identifier, if specified, is written to the key
         :returns: `!True` if key has been created successfully.
 
         this method should create atomically initialize key and return `!True`
@@ -188,5 +266,5 @@ class AbstractDCS:
         :param timeout: timeout in seconds
         :returns: `!True` if you would like to reschedule the next run of ha cycle"""
 
-        sleep(timeout)
+        self.event.wait(timeout)
-        return False
+        return self.event.isSet()

patroni/etcd.py

@@ -10,7 +10,8 @@ import urllib3
 
 from dns.exception import DNSException
 from dns import resolver
-from patroni.dcs import AbstractDCS, Cluster, DCSError, 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
 
@@ -51,7 +52,11 @@ class Client(etcd.Client):
     def api_execute(self, path, method, **kwargs):
         # Update machines_cache if previous attempt of update has failed
         self._update_machines_cache and self._load_machines_cache()
+        try:
             return super(Client, self).api_execute(path, method, **kwargs)
+        except etcd.EtcdConnectionFailed:
+            self._update_machines_cache = True
+            raise
 
     @staticmethod
     def get_srv_record(host):
@@ -80,7 +85,7 @@ class Client(etcd.Client):
         for host, port in self.get_srv_record(discovery_srv):
             url = '{}://{}:{}/members'.format(self._protocol, host, port)
             try:
-                response = requests.get(url)
+                response = requests.get(url, timeout=5)
                 if response.ok:
                     for member in response.json():
                         ret.extend(member['clientURLs'])
@@ -146,14 +151,12 @@ 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,
                                               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)
@@ -170,16 +173,16 @@ class Etcd(AbstractDCS):
 
     @staticmethod
     def member(node):
-        conn_url, api_url = parse_connection_string(node.value)
+        return Member.from_node(node.modifiedIndex, os.path.basename(node.key), node.ttl, node.value)
-        return Member(node.modifiedIndex, os.path.basename(node.key), conn_url, api_url, node.expiration, node.ttl)
 
-    def get_cluster(self):
+    def _load_cluster(self):
         try:
             result = self.retry(self.client.read, self.client_path(''), recursive=True)
             nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves}
 
             # get initialize flag
-            initialize = bool(nodes.get(self._INITIALIZE, False))
+            initialize = nodes.get(self._INITIALIZE, None)
+            initialize = initialize and initialize.value
 
             # get last leader operation
             last_leader_operation = nodes.get(self._LEADER_OPTIME, None)
@@ -191,22 +194,25 @@ 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)
 
-            self.cluster = Cluster(initialize, leader, last_leader_operation, members)
+            # failover key
+            failover = nodes.get(self._FAILOVER, None)
+            if failover:
+                failover = Failover.from_node(failover.modifiedIndex, failover.value)
+
+            self._cluster = Cluster(initialize, leader, last_leader_operation, members, failover)
         except etcd.EtcdKeyNotFound:
-            self.cluster = Cluster(False, None, None, [])
+            self._cluster = Cluster(False, None, None, [], None)
         except:
-            self.cluster = None
             logger.exception('get_cluster')
             raise EtcdError('Etcd is not responding properly')
-        return self.cluster
 
     @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):
@@ -222,18 +228,20 @@ class Etcd(AbstractDCS):
         return False
 
     @catch_etcd_errors
-    def write_leader_optime(self, state_handler):
+    def set_failover_value(self, value, index=None):
-        return self.client.set(self.leader_optime_path, state_handler.last_operation())
+        return self.client.write(self.failover_path, value, prevIndex=index or 0)
 
     @catch_etcd_errors
-    def update_leader(self, state_handler):
+    def write_leader_optime(self, last_operation):
-        ret = self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
+        return self.client.set(self.leader_optime_path, last_operation)
-        ret and self.write_leader_optime(state_handler)
-        return ret
 
     @catch_etcd_errors
-    def initialize(self):
+    def update_leader(self):
-        return self.client.write(self.initialize_path, self._name, prevExist=False)
+        return self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
+
+    @catch_etcd_errors
+    def initialize(self, create_new=True, sysid=""):
+        return self.retry(self.client.write, self.initialize_path, sysid, prevExist=(not create_new))
 
     @catch_etcd_errors
     def delete_leader(self):
@@ -241,13 +249,14 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def cancel_initialization(self):
-        return self.client.delete(self.initialize_path, prevValue=self._name)
+        return self.retry(self.client.delete, self.initialize_path, prevValue=self._name)
 
     def watch(self, timeout):
+        cluster = self.cluster
         # watch on leader key changes if it is defined and current node is not lock owner
-        if self.cluster and self.cluster.leader and self.cluster.leader.name != self._name:
+        if cluster and cluster.leader and cluster.leader.name != self._name:
             end_time = time.time() + timeout
-            index = self.cluster.leader.index
+            index = cluster.leader.index
 
             while index and timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
                 try:
@@ -263,4 +272,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

@@ -1,26 +1,31 @@
+import json
 import logging
 import psycopg2
+import requests
+import sys
 
+from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
+from multiprocessing.pool import ThreadPool
 
 logger = logging.getLogger(__name__)
 
 
 class Ha:
 
-    def __init__(self, state_handler, etcd):
+    def __init__(self, patroni):
-        self.state_handler = state_handler
+        self.patroni = patroni
-        self.dcs = etcd
+        self.state_handler = patroni.postgresql
+        self.dcs = patroni.dcs
         self.cluster = None
         self.old_cluster = None
+        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():
+        if not cluster.is_unlocked() or not self.old_cluster:
-            self.old_cluster = self.cluster
-        if not self.old_cluster:
             self.old_cluster = cluster
         self.cluster = cluster
 
@@ -28,26 +33,51 @@ class Ha:
         return self.dcs.attempt_to_acquire_leader()
 
     def update_lock(self):
-        return self.dcs.update_leader(self.state_handler)
+        ret = self.dcs.update_leader()
+        if ret:
+            try:
+                self.dcs.write_leader_optime(self.state_handler.last_operation())
+            except:
+                pass
+        return ret
 
     def has_lock(self):
         lock_owner = self.cluster.leader and self.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 bootstrap(self):
+    def touch_member(self):
-        if not self.cluster.is_unlocked():  # cluster already has leader
+        data = {
-            logger.info('trying to bootstrap from leader', )
+            'conn_url': self.state_handler.connection_string,
-            if self.state_handler.bootstrap(self.cluster.leader):
+            'api_url': self.patroni.api.connection_string,
-                return 'bootstrapped from leader'
+            '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, leader):
+        if self.state_handler.bootstrap(leader):
+            logger.info('bootstrapped from leader')
         else:
             self.state_handler.stop('immediate')
             self.state_handler.remove_data_directory()
-                return 'failed to bootstrap from leader'
+            logger.error('failed to bootstrap from leader')
+
+    def bootstrap(self):
+        if not self.cluster.is_unlocked():  # cluster already has leader
+            self._async_executor.schedule('bootstrap from leader')
+            self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, ))
+            return 'trying to bootstrap from leader'
         elif not self.cluster.initialize:  # no initialize key
-            if self.dcs.initialize():  # race for initialization
+            if self.dcs.initialize(create_new=True):  # race for initialization
                 try:
                     self.state_handler.bootstrap()
+                    self.dcs.initialize(create_new=False, sysid=self.state_handler.sysid)
                 except:  # initdb or start failed
                     # remove initialization key and give a chance to other members
                     logger.info("removing initialize key after failed attempt to initialize the cluster")
@@ -63,20 +93,38 @@ class Ha:
             return 'waiting for leader to bootstrap'
 
     def recover(self):
-        if self.state_handler.is_healthy():
-            return False
         has_lock = self.has_lock()
-        self.state_handler.write_recovery_conf(None if has_lock else self.cluster.leader)
+
-        self.state_handler.start()
+        # try to see if we are the former master that crashed. If so - we likely need to run pg_rewind
-        if has_lock:
+        # in order to join the former standby being promoted.
+        pg_controldata = self.state_handler.controldata()
+        if not has_lock and pg_controldata and\
+                pg_controldata.get('Database cluster state', '') == 'in production':  # crashed master
+            self.state_handler.require_rewind()
+
+        # XXX: follow the leader calls stop, which might take quite some time.
+        # perhaps we should run sync asynchronously
+        # (we still need the exit code from follow_the_leader)
+        ret = self.state_handler.follow_the_leader(None if has_lock else self.cluster.leader, recovery=True)
+        if not ret:
+            if not has_lock:
+                return 'failed to start postgres'
+            self.dcs.delete_leader()
+            self.dcs.reset_cluster()
+            return 'removed leader key after trying and failing to start postgres'
+        if not has_lock:
+            return 'started as a secondary'
         logger.info('started as readonly because i had the session lock')
         self.load_cluster_from_dcs()
-        return True
 
     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
-        self.state_handler.follow_the_leader(self.cluster.leader)
+        leader = self.cluster.leader
+        leader = None if (leader and leader.name) == self.state_handler.name else leader
+        if not self.state_handler.check_recovery_conf(leader):
+            self._async_executor.schedule('changing primary_conninfo and restarting')
+            self._async_executor.run_async(self.state_handler.follow_the_leader, (leader, ))
         return ret
 
     def enforce_master_role(self, message, promote_message):
@@ -86,9 +134,145 @@ class Ha:
             self.state_handler.promote()
             return promote_message
 
+    @staticmethod
+    def fetch_node_status(member):
+        """This function perform http get request on member.api_url and fetches its status
+        :returns: tuple(`member`, reachable, in_recovery, xlog_location)
+
+        reachable - `!False` if the node is not reachable or is not responding with correct JSON
+        in_recovery - `!True` if pg_is_in_recovery() == true
+        xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role."""
+
+        try:
+            response = requests.get(member.api_url, timeout=2, verify=False)
+            logger.info('Got response from %s %s: %s', member.name, member.api_url, response.content)
+            json = response.json()
+            is_master = json['role'] == 'master'
+            xlog_location = json['xlog']['location' if is_master else 'replayed_location']
+            return (member, True, not is_master, xlog_location)
+        except:
+            logging.exception('request failed: GET %s', member.api_url)
+        return (member, False, None, 0)
+
+    def fetch_nodes_statuses(self, members):
+        pool = ThreadPool(len(members))
+        results = pool.map(self.fetch_node_status, members)  # Run API calls on members in parallel
+        pool.close()
+        pool.join()
+        return results
+
+    def _is_healthiest_node(self, members, check_replication_lag=True):
+        """This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
+
+        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):
+            return False  # Too far behind last reported xlog location on master
+
+        # Prepare list of nodes to run check against
+        members = [m for m in members if m.name != self.state_handler.name and m.api_url]
+
+        if members:
+            my_xlog_location = self.state_handler.xlog_position()
+            for member, reachable, in_recovery, xlog_location in self.fetch_nodes_statuses(members):
+                if reachable:  # If the node is unreachable it's not healhy
+                    if not in_recovery:
+                        logger.warning('Master (%s) is still alive', member.name)
+                        return False
+                    if my_xlog_location < xlog_location:
+                        return False
+        return True
+
+    def is_failover_possible(self, members):
+        ret = False
+        members = [m for m in members if m.name != self.state_handler.name and m.api_url]
+        if members:
+            for member, reachable, in_recovery, xlog_location in self.fetch_nodes_statuses(members):
+                if reachable:
+                    ret = True  # TODO: check xlog_location
+                else:
+                    logger.info('Member %s is not reachable', member.name)
+        else:
+            logger.warning('manual failover: members list is empty')
+        return ret
+
+    def manual_failover_process_no_leader(self):
+        failover = self.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]
+            if members:
+                member, reachable, in_recovery, xlog_location = self.fetch_node_status(members[0])
+                if reachable:  # node is healthy
+                    logger.info('manual failover: to %s, i am %s', member.name, self.state_handler.name)
+                    return False
+                # we wanted to failover to specific member but it is not healthy
+                logger.warning('manual failover: member %s is unhealthy', member.name)
+
+            # at this point we should consider all members as a candidates for failover
+            # i.e. we assume that failover.member is None
+
+        # try to pick some other members to failover and check that they are healthy
+        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]
+                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
+                    return True
+
+            # 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]
+        return self._is_healthiest_node(members, check_replication_lag=False)
+
+    def is_healthiest_node(self):
+        if self.cluster.failover:
+            return self.manual_failover_process_no_leader()
+
+        # run usual health check
+        members = {m.name: m for m in self.cluster.members + self.old_cluster.members}
+        return self._is_healthiest_node(members.values())
+
+    def demote(self, delete_leader=True):
+        if delete_leader:
+            self.state_handler.stop()
+            self.dcs.delete_leader()
+            self.dcs.reset_cluster()
+        self.state_handler.follow_the_leader(None)
+
+    def process_manual_failover_from_leader(self):
+        failover = self.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]
+                if self.is_failover_possible(members):  # check that there are healthy members
+                    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')
+            else:
+                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)
+
+        logger.info('Trying to clean up failover key')
+        self.dcs.manual_failover('', '', self.cluster.failover.index)
+
     def process_unhealthy_cluster(self):
-        if self.state_handler.is_healthiest_node(self.old_cluster):
+        if self.is_healthiest_node():
             if self.acquire_lock():
+                if self.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',
                                                 'promoted self to leader by acquiring session lock')
             else:
@@ -100,6 +284,11 @@ class Ha:
 
     def process_healthy_cluster(self):
         if self.has_lock():
+            if self.cluster.failover:
+                msg = self.process_manual_failover_from_leader()
+                if msg is not None:
+                    return msg
+
             if self.update_lock():
                 return self.enforce_master_role('no action.  i am the leader with the lock',
                                                 'promoted self to leader because i had the session lock')
@@ -112,34 +301,114 @@ 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 run_cycle(self):
+    def schedule(self, action):
+        with self._async_executor:
+            return self._async_executor.schedule(action)
+
+    def restart_scheduled(self):
+        return self._async_executor.scheduled_action == 'restart'
+
+    def schedule_reinitialize(self):
+        return self.schedule('reinitialize')
+
+    def reinitialize_scheduled(self):
+        return self._async_executor.scheduled_action == 'reinitialize'
+
+    def restart(self):
+        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')
+
+    def reinitialize(self, cluster):
+        self.state_handler.stop('immediate')
+        self.state_handler.remove_data_directory()
+        self.copy_backup_from_leader(cluster.leader)
+
+    def process_scheduled_action(self):
+        if self.reinitialize_scheduled():
+            if self.cluster.is_unlocked():
+                logger.error('Cluster has no leader, can not reinitialize')
+                self._async_executor.reset_scheduled_action()
+            elif self.has_lock():
+                logger.error('I am the leader, can not reinitialize')
+                self._async_executor.reset_scheduled_action()
+            else:
+                self._async_executor.run_async(self.reinitialize, args=(self.cluster, ))
+                return 'reinitialize started'
+
+    def handle_long_action_in_progress(self):
+        if self.has_lock():
+            if self.update_lock():
+                return 'updated leader lock during ' + self._async_executor.scheduled_action
+            else:
+                return 'failed to update leader lock during ' + self._async_executor.scheduled_action
+        elif self.cluster.is_unlocked():
+            return 'not healthy enough for leader race'
+        else:
+            return self._async_executor.scheduled_action + ' in progress'
+
+    def sysid_valid(self, sysid):
+        # sysid does tv_sec << 32, where tv_sec is the number of seconds sine 1970,
+        # so even 1 << 32 would have 10 digits.
+        return str(sysid) and len(str(sysid)) >= 10 and str(sysid).isdigit()
+
+    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.cluster.is_unlocked() and not self.sysid_valid(self.cluster.initialize) and self.has_lock():
-                self.dcs.initialize()  # fix it
+                self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid)
+
+            if self._async_executor.busy:
+                return self.handle_long_action_in_progress()
+
+            # currently it can trigger only reinitialize
+            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.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked():
-                self.dcs.initialize()
+                self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid)
+            else:
+                # check if we are allowed to join
+                if self.sysid_valid(self.cluster.initialize) and self.cluster.initialize != self.state_handler.sysid:
+                    logger.fatal("system ID mismatch, node {0} belongs to a different cluster".
+                                 format(self.state_handler.name))
+                    sys.exit(1)
 
             # try to start dead postgres
-            if self.recover() and not self.has_lock():
+            if not self.state_handler.is_healthy():
-                # no lock, do not try to promote immediately
+                msg = self.recover()
-                return 'started as a secondary'
+                if msg is not None:
+                    return msg
 
+            try:
                 if self.cluster.is_unlocked():
                     return self.process_unhealthy_cluster()
                 else:
                     return self.process_healthy_cluster()
+            finally:
+                self.state_handler.sync_replication_slots(self.cluster)
         except DCSError:
             logger.error('Error communicating with DCS')
-            if self.state_handler.is_leader():
+            if self.state_handler.is_running() and self.state_handler.is_leader():
-                self.state_handler.demote(None)
+                self.demote(delete_leader=False)
                 return 'demoted self because DCS is not accessible and i was a leader'
         except (psycopg2.Error, PostgresConnectionException):
-            logger.exception('Error communicating with Postgresql.  Will try again')
+            logger.exception('Error communicating with Postgresql. Will try again later')
+
+    def run_cycle(self):
+        with self._async_executor:
+            return self._run_cycle()

patroni/postgresql.py

@@ -9,6 +9,7 @@ import time
 from patroni.exceptions import PostgresConnectionException, PostgresException
 from patroni.utils import Retry, RetryFailedError
 from six.moves.urllib_parse import urlparse
+from threading import Lock
 
 logger = logging.getLogger(__name__)
 
@@ -47,6 +48,8 @@ class Postgresql:
         self.replication = config['replication']
         self.superuser = config['superuser']
         self.admin = config['admin']
+        self.pgpass = config.get('pgpass', None) or os.path.join(os.path.expanduser('~'), 'pgpass')
+        self.pg_rewind = config.get('pg_rewind', {})
         self.callback = config.get('callbacks', {})
         self.use_slots = config.get('use_slots', True)
         self.schedule_load_slots = self.use_slots
@@ -56,7 +59,6 @@ class Postgresql:
         self.postmaster_pid = os.path.join(self.data_dir, 'postmaster.pid')
         self.trigger_file = config.get('recovery_conf', {}).get('trigger_file', None) or 'promote'
         self.trigger_file = os.path.abspath(os.path.join(self.data_dir, self.trigger_file))
-        self._role = 'replica'
 
         self._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir]
 
@@ -67,8 +69,50 @@ class Postgresql:
 
         self._connection = None
         self._cursor_holder = None
-        self.members = []  # list of already existing replication slots
+        self._need_rewind = False
-        self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException)
+        self._sysid = None
+        self.replication_slots = []  # list of already existing replication slots
+        self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException)
+
+        self._state = 'stopped'
+        self._state_lock = Lock()
+        self._role = 'replica'
+        self._role_lock = Lock()
+
+        if self.is_running():
+            self._state = 'running'
+            self._role = 'master' if self.is_leader() else 'replica'
+
+    @property
+    def can_rewind(self):
+        """ check if pg_rewind executable is there and that pg_controldata indicates
+            we have either wal_log_hints or checksums turned on
+        """
+        # low-hanging fruit: check if pg_rewind configuration is there
+        if not self.pg_rewind or\
+           not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')):
+            return False
+
+        cmd = ['pg_rewind', '--help']
+        try:
+            ret = subprocess.call(cmd, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
+            if ret != 0:  # pg_rewind is not there, close up the shop and go home
+                return False
+        except OSError:
+            return False
+        # check if the cluster's configuration permits pg_rewind
+        data = self.controldata()
+        return data.get('wal_log_hints setting', 'off') == 'on' or data.get('Data page checksum version', '0') != '0'
+
+    @property
+    def sysid(self):
+        if not self._sysid:
+            data = self.controldata()
+            self._sysid = data.get('Database system identifier', "")
+        return self._sysid
+
+    def require_rewind(self):
+        self._need_rewind = True
 
     def get_local_address(self):
         listen_addresses = self.listen_addresses.split(',')
@@ -89,9 +133,15 @@ class Postgresql:
 
     def _cursor(self):
         if not self._cursor_holder or self._cursor_holder.closed or self._cursor_holder.connection.closed != 0:
+            logger.info("established a new patroni connection to the postgres cluster")
             self._cursor_holder = self.connection().cursor()
         return self._cursor_holder
 
+    def close_connection(self):
+        if self._cursor_holder and self._cursor_holder.connection and self._cursor_holder.connection.closed == 0:
+            self._cursor_holder.connection.close()
+            logger.info("closed patroni connection to the postgresql cluster")
+
     def _query(self, sql, *params):
         cursor = None
         try:
@@ -101,6 +151,8 @@ class Postgresql:
         except psycopg2.Error as e:
             if cursor and cursor.connection.closed == 0:
                 raise e
+            if self.state == 'restarting':
+                raise RetryFailedError('cluster is being restarted')
             raise PostgresConnectionException('connection problems')
 
     def query(self, sql, *params):
@@ -113,23 +165,30 @@ class Postgresql:
         return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []
 
     def initialize(self):
+        self.set_state('initalizing new cluster')
         ret = subprocess.call(self._pg_ctl + ['initdb', '-o', '--encoding=UTF8']) == 0
-        ret and self.write_pg_hba()
+        if ret:
+            self.write_pg_hba()
+        else:
+            self.set_state('initdb failed')
         return ret
 
     def delete_trigger_file(self):
         os.path.exists(self.trigger_file) and os.unlink(self.trigger_file)
 
+    def write_pgpass(self, record):
+        with open(self.pgpass, 'w') as f:
+            os.fchmod(f.fileno(), 0o600)
+            f.write('{host}:{port}:*:{user}:{password}\n'.format(**record))
+
+        env = os.environ.copy()
+        env['PGPASSFILE'] = self.pgpass
+        return env
+
     def sync_from_leader(self, leader):
         r = parseurl(leader.conn_url)
 
-        pgpass = 'pgpass'
+        env = self.write_pgpass(r)
-        with open(pgpass, 'w') as f:
-            os.fchmod(f.fileno(), 0o600)
-            f.write('{host}:{port}:*:{user}:{password}\n'.format(**r))
-
-        env = os.environ.copy()
-        env['PGPASSFILE'] = pgpass
         return self.create_replica(r, env) == 0
 
     @staticmethod
@@ -213,40 +272,82 @@ class Postgresql:
 
     @property
     def role(self):
+        with self._role_lock:
             return self._role
 
+    def set_role(self, value):
+        with self._role_lock:
+            self._role = value
+
+    @property
+    def state(self):
+        with self._state_lock:
+            return self._state
+
+    def set_state(self, value):
+        with self._state_lock:
+            self._state = value
+
     def start(self, block_callbacks=False):
         if self.is_running():
-            self._role = 'master' if self.is_leader() else 'replica'
-            self.schedule_load_slots = self.use_slots
             logger.error('Cannot start PostgreSQL because one is already running.')
-            return False
+            return True
 
-        self._role = 'replica' if os.path.exists(self.recovery_conf) else 'master'
+        self.set_role('replica' if os.path.exists(self.recovery_conf) else 'master')
         if os.path.exists(self.postmaster_pid):
             os.remove(self.postmaster_pid)
             logger.info('Removed %s', self.postmaster_pid)
 
+        if not block_callbacks:
+            self.set_state('starting')
+
         ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0
+
+        self.set_state('running' if ret else 'start failed')
+
         self.schedule_load_slots = ret and self.use_slots
         self.save_configuration_files()
         # block_callbacks is used during restart to avoid
         # running start/stop callbacks in addition to restart ones
-        ret and not block_callbacks and ret and self.call_nowait(ACTION_ON_START)
+        ret and not block_callbacks and self.call_nowait(ACTION_ON_START)
         return ret
 
-    def stop(self, mode='fast', block_callbacks=False):
+    def checkpoint(self):
-        if block_callbacks:
         try:
-                self.query('SET statement_timeout TO 0')
+            r = parseurl('postgres://{}/postgres'.format(self.local_address))
-                self.query('CHECKPOINT')
+            r['options'] = '-c statement_timeout=0'
+            with psycopg2.connect(**r) as conn:
+                conn.autocommit = True
+                with conn.cursor() as cur:
+                    cur.execute('CHECKPOINT')
         except:
-                logging.exception('Exception diring CHECKPOINT')
+            logging.exception('Exception during CHECKPOINT')
+
+    def stop(self, mode='fast', block_callbacks=False):
+        # make sure we close all connections established against
+        # the former node, otherwise, we might get a stalled one
+        # after kill -9, which would report incorrect data to
+        # patroni.
+
+        self.close_connection()
+        if not self.is_running():
+            if not block_callbacks:
+                self.set_state('stopped')
+            return True
+
+        if block_callbacks:
+            self.checkpoint()
+        else:
+            self.set_state('stopping')
 
         ret = subprocess.call(self._pg_ctl + ['stop', '-m', mode]) == 0
         # block_callbacks is used during restart to avoid
         # running start/stop callbacks in addition to restart ones
-        ret and not block_callbacks and self.call_nowait(ACTION_ON_STOP)
+        if not ret:
+            self.set_state('stop failed')
+        elif not block_callbacks:
+            self.set_state('stopped')
+            self.call_nowait(ACTION_ON_STOP)
         return ret
 
     def reload(self):
@@ -255,8 +356,12 @@ class Postgresql:
         return ret
 
     def restart(self):
+        self.set_state('restarting')
         ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True)
-        ret and self.call_nowait(ACTION_ON_RESTART)
+        if ret:
+            self.call_nowait(ACTION_ON_RESTART)
+        else:
+            self.set_state('restart failed ({})'.format(self.state))
         return ret
 
     def server_options(self):
@@ -271,36 +376,9 @@ class Postgresql:
             return False
         return True
 
-    def is_healthiest_node(self, cluster):
+    def check_replication_lag(self, last_leader_operation):
-        if self.is_leader():
+        return (last_leader_operation if last_leader_operation else 0) - self.xlog_position() <=\
-            return True
+            self.config.get('maximum_lag_on_failover', 0)
-
-        if cluster.last_leader_operation - self.xlog_position() > self.config.get('maximum_lag_on_failover', 0):
-            return False
-
-        for member in cluster.members:
-            if member.name == self.name:
-                continue
-            try:
-                r = parseurl(member.conn_url)
-                member_conn = psycopg2.connect(**r)
-                member_conn.autocommit = True
-                member_cursor = member_conn.cursor()
-                member_cursor.execute(
-                    "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')",
-                    (self.xlog_position(),))
-                row = member_cursor.fetchone()
-                member_cursor.close()
-                member_conn.close()
-                logger.error([self.name, member.name, row])
-                if not row[0]:
-                    logger.warning('Master (%s) is still alive', member.name)
-                    return False
-                if row[1] < 0:
-                    return False
-            except psycopg2.Error:
-                continue
-        return True
 
     def write_pg_hba(self):
         with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f:
@@ -324,10 +402,7 @@ class Postgresql:
         with open(self.recovery_conf, 'r') as f:
             for line in f:
                 if line.startswith('primary_conninfo'):
-                    if not pattern:
+                    return pattern and (pattern in line)
-                        return False
-                    return pattern in line
-
         return not pattern
 
     def write_recovery_conf(self, leader):
@@ -342,40 +417,154 @@ recovery_target_timeline = 'latest'
                 for name, value in self.config.get('recovery_conf', {}).items():
                     f.write("{} = '{}'\n".format(name, value))
 
-    def follow_the_leader(self, leader):
+    def rewind(self, leader):
-        if not self.check_recovery_conf(leader):
+        # prepare pg_rewind connection
+        r = parseurl(leader.conn_url)
+        r.update(self.pg_rewind)
+        r['user'] = r['username']
+        env = self.write_pgpass(r)
+        pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
+        logger.info("running pg_rewind from {}".format(pc))
+        pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
+        try:
+            ret = (subprocess.call(pg_rewind, env=env) == 0)
+        except:
+            ret = False
+        if ret:
             self.write_recovery_conf(leader)
-            run_callback = self.role == 'master'
+        return ret
-            self.restart()
+
-            run_callback and self.call_nowait(ACTION_ON_ROLE_CHANGE)
+    def controldata(self):
+        """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """
+        result = {}
+        try:
+            data = subprocess.check_output(['pg_controldata', self.data_dir])
+            if data:
+                data = data.decode().splitlines()
+                result = {l.split(':')[0].replace('Current ', '', 1): l.split(':')[1].strip() for l in data if l}
+        except subprocess.CalledProcessError:
+            logger.exception("Error when calling pg_controldata")
+        finally:
+            return result
+
+    def read_postmaster_opts(self):
+        """ returns the list of option names/values from postgres.opts, Empty dict if read failed or no file """
+        result = {}
+        try:
+            with open(os.path.join(self.data_dir, "postmaster.opts")) as f:
+                data = f.read()
+                opts = [opt.strip('"\n') for opt in data.split(' "')]
+                for opt in opts:
+                    if '=' in opt and opt.startswith('--'):
+                        name, val = opt.split('=', 1)
+                        name = name.strip('-')
+                        result[name] = val
+        except IOError:
+            logger.exception('Error when reading postmaster.opts')
+        finally:
+            return result
+
+    def single_user_mode(self, command=None, options={}):
+        """ run a given command in a single-user mode. If the command is empty - then just start and stop """
+        cmd = ['postgres', '--single', '-D', self.data_dir]
+        for opt in sorted(options):
+            cmd.extend(['-c', '{0}={1}'.format(opt, options[opt])])
+        # need a database name to connect
+        cmd.append('postgres')
+        p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
+        if p:
+            command and p.communicate('{}\n'.format(command))
+            p.stdin.close()
+            return p.wait()
+        return 1
+
+    def cleanup_archive_status(self):
+        status_dir = os.path.join(self.data_dir, 'pg_xlog', 'archive_status')
+        if os.path.isdir(status_dir):
+            for f in os.listdir(status_dir):
+                path = os.path.join(status_dir, f)
+                try:
+                    if os.path.islink(path):
+                        os.unlink(path)
+                    elif os.path.isfile(path):
+                        os.remove(path)
+                except:
+                    logger.exception("Unable to remove {}".format(path))
+
+    def follow_the_leader(self, leader, recovery=False):
+        if not self.check_recovery_conf(leader) or recovery:
+            change_role = (self.role == 'master')
+
+            self._need_rewind = (self._need_rewind or change_role) and self.can_rewind
+            if self._need_rewind:
+                logger.info("set the rewind flag after demote")
+            self.write_recovery_conf(leader)
+            if not leader or not self._need_rewind:  # do not rewind until the leader becomes available
+                ret = self.restart()
+            else:  # we have a leader and need to rewind
+                if self.is_running():
+                    self.stop()
+                # at present, pg_rewind only runs when the cluster is shut down cleanly
+                # and not shutdown in recovery. We have to remove the recovery.conf if present
+                # and start/shutdown in a single user mode to emulate this.
+                # XXX: if recovery.conf is linked, it will be written anew as a normal file.
+                if os.path.islink(self.recovery_conf):
+                    os.unlink(self.recovery_conf)
+                else:
+                    os.remove(self.recovery_conf)
+                # Archived segments might be useful to pg_rewind,
+                # clean the flags that tell we should remove them.
+                self.cleanup_archive_status()
+                # Start in a single user mode and stop to produce a clean shutdown
+                opts = self.read_postmaster_opts()
+                opts['archive_mode'] = 'on'
+                opts['archive_command'] = 'false'
+                self.single_user_mode(options=opts)
+                if self.rewind(leader):
+                    ret = self.start()
+                else:
+                    logger.error("unable to rewind the former master")
+                    self.remove_data_directory()
+                    ret = True
+                self._need_rewind = False
+            change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE)
+            return ret
+        else:
+            return True
 
     def save_configuration_files(self):
         """
-            copy postgresql.conf to postgresql.conf.backup to preserve it in the WAL-e backup.
+            copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files
-            see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239
+            - originally stored as symlinks, those are normally skipped by pg_basebackup
+            - in case of WAL-E basebackup (see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239)
         """
+        try:
             for f in self.configuration_to_save:
-            shutil.copy(f, f + '.backup')
+                os.path.isfile(f) and shutil.copy(f, f + '.backup')
+        except:
+            logger.exception('unable to create backup copies of configuration files')
 
     def restore_configuration_files(self):
         """ restore a previously saved postgresql.conf """
         try:
             for f in self.configuration_to_save:
-                shutil.copy(f + '.backup', f)
+                not os.path.isfile(f) and os.path.isfile(f+'.backup') and shutil.copy(f + '.backup', f)
         except:
-            logger.exception('unable to restore configuration from WAL-E backup')
+            logger.exception('unable to restore configuration files from backup')
 
     def promote(self):
         if self.role == 'master':
             return True
         ret = subprocess.call(self._pg_ctl + ['promote']) == 0
         if ret:
-            self._role = 'master'
+            self.set_role('master')
+            logger.info("cleared rewind flag after becoming the leader")
+            self._need_rewind = False
             self.call_nowait(ACTION_ON_ROLE_CHANGE)
         return ret
 
-    def demote(self, leader):
+    def demote(self):
-        self.follow_the_leader(leader)
+        self.follow_the_leader(None)
 
     def create_replication_user(self):
         self.query('CREATE USER "{}" WITH REPLICATION ENCRYPTED PASSWORD %s'.format(
@@ -397,31 +586,34 @@ 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:
             cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")
-            self.members = [r[0] for r in cursor]
+            self.replication_slots = [r[0] for r in cursor]
             self.schedule_load_slots = False
 
     def sync_replication_slots(self, cluster):
         if self.use_slots:
+            try:
                 self.load_replication_slots()
-            members = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else []
+                slots = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else []
                 # drop unused slots
-            for slot in set(self.members) - set(members):
+                for slot in set(self.replication_slots) - set(slots):
                     self.query("""SELECT pg_drop_replication_slot(%s)
                                    WHERE EXISTS(SELECT 1 FROM pg_replication_slots
                                    WHERE slot_name = %s)""", slot, slot)
 
                 # create new slots
-            for slot in set(members) - set(self.members):
+                for slot in set(slots) - set(self.replication_slots):
                     self.query("""SELECT pg_create_physical_replication_slot(%s)
                                    WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots
                                    WHERE slot_name = %s)""", slot, slot)
 
-            self.members = members
+                self.replication_slots = slots
+            except:
+                logger.exception('Exception when changing replication slots')
 
     def last_operation(self):
         return str(self.xlog_position())
@@ -446,6 +638,7 @@ recovery_target_timeline = 'latest'
                 raise PostgresException("Could not bootstrap master PostgreSQL")
         else:
             if self.sync_from_leader(current_leader):
+                self.restore_configuration_files()
                 self.write_recovery_conf(current_leader)
                 ret = self.start()
         return ret

patroni/scripts/aws.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python
 
 import logging
 import requests

patroni/scripts/restore.py

@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python
 # arguments are:
 #   - cluster scope
 #   - cluster role

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,8 @@ import time
 
 from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
-from patroni.dcs import AbstractDCS, Cluster, DCSError, 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
 
@@ -90,9 +91,8 @@ 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.cluster = None
+        self._my_member_data = None
         self.fetch_cluster = True
         self.last_leader_operation = 0
 
@@ -104,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:
@@ -115,8 +115,7 @@ class ZooKeeper(AbstractDCS):
 
     @staticmethod
     def member(name, value, znode):
-        conn_url, api_url = parse_connection_string(value)
+        return Member.from_node(znode.version, name, znode.ephemeralOwner, value)
-        return Member(znode.version, name, conn_url, api_url, None, None)
 
     def get_children(self, key, watch=None):
         try:
@@ -133,17 +132,20 @@ class ZooKeeper(AbstractDCS):
         return members
 
     def _inner_load_cluster(self):
-        self.cluster_event.clear()
+        self.fetch_cluster = False
-        nodes = set(self.get_children(self.client_path('')))
+        self.event.clear()
+        nodes = set(self.get_children(self.client_path(''), self.cluster_watcher))
+        if not nodes:
+            self.fetch_cluster = True
 
         # get initialize flag
-        initialize = self._INITIALIZE in nodes
+        initialize = (self.get_node(self.initialize_path) or [None])[0] if self._INITIALIZE in nodes else None
 
         # get list of members
         members = self.load_members() if self._MEMBERS[:-1] in nodes else []
 
         # get leader
-        leader = self.get_node(self.leader_path, self.cluster_watcher) if self._LEADER in nodes else None
+        leader = self.get_node(self.leader_path) if self._LEADER in nodes else None
         if leader:
             client_id = self.client.client_id
             if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner:
@@ -152,17 +154,22 @@ 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
 
-        # get last leader operation
+        # failover key
-        self.last_leader_operation = self.get_node(self.leader_optime_path) if self.fetch_cluster else None
+        failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
-        self.last_leader_operation = 0 if self.last_leader_operation is None else int(self.last_leader_operation[0])
+        if failover:
-        self.cluster = Cluster(initialize, leader, self.last_leader_operation, members)
+            failover = Failover.from_node(failover[1].version, failover[0])
 
-    def get_cluster(self):
+        # get last leader operation
+        optime = self.get_node(self.leader_optime_path) if self._OPTIME in nodes and self.fetch_cluster else None
+        self.last_leader_operation = 0 if optime is None else int(optime[0])
+        self._cluster = Cluster(initialize, leader, self.last_leader_operation, members, failover)
+
+    def _load_cluster(self):
         if self.exhibitor and self.exhibitor.poll():
             self.client.set_hosts(self.exhibitor.zookeeper_hosts)
 
@@ -170,11 +177,9 @@ class ZooKeeper(AbstractDCS):
             try:
                 self.client.retry(self._inner_load_cluster)
             except:
-                self.cluster = None
                 logger.exception('get_cluster')
                 self.session_listener(KazooState.LOST)
                 raise ZooKeeperError('ZooKeeper in not responding properly')
-        return self.cluster
 
     def _create(self, path, value, **kwargs):
         try:
@@ -188,31 +193,61 @@ class ZooKeeper(AbstractDCS):
         ret or logger.info('Could not take out TTL lock')
         return ret
 
-    def initialize(self):
+    def set_failover_value(self, value, index=None):
-        return self._create(self.initialize_path, self._name, makepath=True)
-
-    def touch_member(self, connection_string, ttl=None):
-        if self.cluster and any(m.name == self._name for m in self.cluster.members):
-            return True
-        path = self.member_path
-        connection_string = connection_string.encode('utf-8')
         try:
-            self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True)
+            self.client.retry(self.client.set, self.failover_path, value.encode('utf-8'), version=index or -1)
+            return True
+        except NoNodeError:
+            return value == '' or (not index and self._create(self.failover_path, value))
+        except:
+            logging.exception('set_failover_value')
+            return False
+
+    def initialize(self, create_new=True, sysid=""):
+        return self._create(self.initialize_path, sysid, makepath=True) if create_new \
+            else self.client.retry(self.client.set, self.initialize_path,  sysid.encode("utf-8"))
+
+    def touch_member(self, data, ttl=None):
+        cluster = self.cluster
+        me = cluster and ([m for m in cluster.members if m.name == self._name] or [None])[0]
+        path = self.member_path
+        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:
+            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:
-                self.client.retry(self.client.delete, path)
+                self.client.retry(self.client.set, path, data)
-                self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True)
+                self._my_member_data = data
                 return True
             except:
                 logger.exception('touch_member')
+        except:
+            logger.exception('touch_member')
         return False
 
     def take_leader(self):
         return self.attempt_to_acquire_leader()
 
-    def update_leader(self, state_handler):
+    def write_leader_optime(self, last_operation):
-        last_operation = state_handler.last_operation().encode('utf-8')
+        last_operation = last_operation.encode('utf-8')
         if last_operation != self.last_leader_operation:
             self.last_leader_operation = last_operation
             path = self.leader_optime_path
@@ -225,11 +260,14 @@ class ZooKeeper(AbstractDCS):
                     logger.exception('Failed to create %s', path)
             except:
                 logger.exception('Failed to update %s', path)
+
+    def update_leader(self):
         return True
 
     def delete_leader(self):
-        if isinstance(self.cluster, Cluster) and self.cluster.leader.name == self._name:
+        self.client.restart()
-            self.client.delete(self.leader_path, version=self.cluster.leader.index)
+        self._my_member_data = None
+        return True
 
     def _cancel_initialization(self):
         node = self.get_node(self.initialize_path)
@@ -243,8 +281,6 @@ class ZooKeeper(AbstractDCS):
             logger.exception("Unable to delete initialize key")
 
     def watch(self, timeout):
-        self.cluster_event.wait(timeout)
+        if super(ZooKeeper, self).watch(timeout):
-        if self.cluster_event.isSet():
             self.fetch_cluster = True
-            return not self.cluster or not self.cluster.leader or self.cluster.leader.name != self._name
+        return self.fetch_cluster
-        return False

postgres0.yml

@@ -34,6 +34,10 @@ postgresql:
   data_dir: data/postgresql0
   maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
   use_slots: True
+  pgpass: /tmp/pgpass0
+  pg_rewind:
+    username: postgres
+    password: zalando
   pg_hba:
   - host all all 0.0.0.0/0 md5
   - hostssl all all 0.0.0.0/0 md5
@@ -42,6 +46,7 @@ postgresql:
     password: rep-pass
     network:  127.0.0.1/32
   superuser:
+    username: postgres
     password: zalando
   admin:
     username: admin
@@ -62,3 +67,4 @@ postgresql:
     archive_timeout: 1800s
     max_replication_slots: 5
     hot_standby: "on"
+    wal_log_hints: "on"

postgres1.yml

@@ -34,6 +34,10 @@ postgresql:
   data_dir: data/postgresql1
   maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
   use_slots: True
+  pgpass: /tmp/pgpass1
+  pg_rewind:
+    username: postgres
+    password: zalando
   pg_hba:
   - host all all 0.0.0.0/0 md5
   - hostssl all all 0.0.0.0/0 md5
@@ -42,6 +46,7 @@ postgresql:
     password: rep-pass
     network: 127.0.0.1/32
   superuser:
+    user: postgres
     password: zalando
   admin:
     username: admin
@@ -62,3 +67,4 @@ postgresql:
     archive_timeout: 1800s
     max_replication_slots: 5
     hot_standby: "on"
+    wal_log_hints: "on"

requirements-py2.txt

@@ -1,7 +1,7 @@
 boto
 dnspython
 mock
-psycopg2
+psycopg2>=2.6.1
 PyYAML
 requests
 six >= 1.7

requirements-py3.txt

@@ -1,7 +1,7 @@
 boto
 mock
 dnspython3
-psycopg2
+psycopg2>=2.6.1
 PyYAML
 requests
 six

tests/test_api.py

@@ -3,6 +3,7 @@ import unittest
 
 from mock import Mock, patch
 from patroni.api import RestApiHandler, RestApiServer
+from patroni.dcs import Member
 from six import BytesIO as IO
 from six.moves import BaseHTTPServer
 from test_postgresql import psycopg2_connect, MockCursor
@@ -10,6 +11,10 @@ from test_postgresql import psycopg2_connect, MockCursor
 
 class MockPostgresql(Mock):
 
+    name = 'test'
+    state = 'running'
+    role = 'master'
+
     def connection(self):
         return psycopg2_connect()
 
@@ -17,9 +22,32 @@ class MockPostgresql(Mock):
         return True
 
 
+class MockHa(Mock):
+
+    dcs = Mock()
+    state_handler = MockPostgresql()
+
+    def schedule_restart(self):
+        return 'restart'
+
+    def schedule_reinitialize(self):
+        return 'reinitialize'
+
+    def restart(self):
+        return (True, '')
+
+    def restart_scheduled(self):
+        return False
+
+    def fetch_nodes_statuses(self, members):
+        return [[None, True, None, None]]
+
+
 class MockPatroni:
 
     postgresql = MockPostgresql()
+    ha = MockHa()
+    dcs = Mock()
 
 
 class MockRequest:
@@ -47,21 +75,77 @@ class MockRestApiServer(RestApiServer):
 class TestRestApiHandler(unittest.TestCase):
 
     def test_do_GET(self):
-        MockRestApiServer(RestApiHandler, b'GET /')
+        MockRestApiServer(RestApiHandler, b'GET /replica')
-        with patch.object(RestApiServer, 'query', Mock(side_effect=psycopg2.OperationalError())):
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})):
-            MockRestApiServer(RestApiHandler, b'GET /')
+            MockRestApiServer(RestApiHandler, b'GET /replica')
-
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):
-    def test_do_GET_sampleauth(self):
+            MockRestApiServer(RestApiHandler, b'GET /replica')
-        MockRestApiServer(RestApiHandler, b'GET /sampleauth')
+        MockRestApiServer(RestApiHandler, b'GET /master')
-        MockRestApiServer(RestApiHandler, b'GET /sampleauth\nAuthorization:')
+        MockPatroni.dcs.cluster.leader.name = MockPostgresql.name
-        MockRestApiServer(RestApiHandler, b'GET /sampleauth\nAuthorization: Basic dGVzdDp0ZXN0')
+        MockRestApiServer(RestApiHandler, b'GET /replica')
+        MockPatroni.dcs.cluster = None
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):
+            MockRestApiServer(RestApiHandler, b'GET /master')
+        with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)):
+            MockRestApiServer(RestApiHandler, b'GET /master')
+        MockRestApiServer(RestApiHandler, b'GET /master')
 
     def test_do_GET_patroni(self):
         MockRestApiServer(RestApiHandler, b'GET /patroni')
 
+    def test_basicauth(self):
+        MockRestApiServer(RestApiHandler, b'POST /restart HTTP/1.0')
+        MockRestApiServer(RestApiHandler, b'POST /restart HTTP/1.0\nAuthorization:')
+
+    def test_do_POST_restart(self):
+        request = b'POST /restart HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0'
+        MockRestApiServer(RestApiHandler, request)
+        with patch.object(MockHa, 'restart', Mock(side_effect=Exception)):
+            MockRestApiServer(RestApiHandler, request)
+
+    @patch.object(MockHa, 'dcs')
+    def test_do_POST_reinitialize(self, dcs):
+        cluster = dcs.get_cluster.return_value
+        request = b'POST /reinitialize HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0'
+        MockRestApiServer(RestApiHandler, request)
+        cluster.is_unlocked.return_value = False
+        MockRestApiServer(RestApiHandler, request)
+        with patch.object(MockHa, 'schedule_reinitialize', Mock(return_value=None)):
+            MockRestApiServer(RestApiHandler, request)
+        cluster.leader.name = 'test'
+        MockRestApiServer(RestApiHandler, request)
+
     @patch('time.sleep', Mock())
     def test_RestApiServer_query(self):
         with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
             MockRestApiServer(RestApiHandler, b'GET /patroni')
         with patch.object(MockPostgresql, 'connection', Mock(side_effect=psycopg2.OperationalError)):
             MockRestApiServer(RestApiHandler, b'GET /patroni')
+
+    @patch('time.sleep', Mock())
+    @patch.object(MockHa, 'dcs')
+    def test_do_POST_failover(self, dcs):
+        cluster = dcs.get_cluster.return_value
+        request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
+                  b'Content-Length: 25\n\n{"leader": "postgresql1"}'
+        MockRestApiServer(RestApiHandler, request)
+        cluster.leader.name = 'postgresql1'
+        MockRestApiServer(RestApiHandler, request)
+        cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'})]
+        MockRestApiServer(RestApiHandler, request)
+        with patch.object(MockPatroni, 'dcs') as d:
+            cluster = d.get_cluster.return_value
+            cluster.leader.name = 'postgresql0'
+            MockRestApiServer(RestApiHandler, request)
+            cluster.leader.name = 'postgresql1'
+            cluster.failover = None
+            MockRestApiServer(RestApiHandler, request)
+            d.get_cluster = Mock(side_effect=Exception())
+            MockRestApiServer(RestApiHandler, request)
+            d.manual_failover.return_value = False
+            MockRestApiServer(RestApiHandler, request)
+        with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])):
+            MockRestApiServer(RestApiHandler, request)
+        request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
+                  b'Content-Length: 50\n\n{"leader": "postgresql1", "member": "postgresql2"}'
+        MockRestApiServer(RestApiHandler, request)

tests/test_async_executor.py

@@ -0,0 +1,18 @@
+import unittest
+
+from mock import Mock, patch
+from patroni.async_executor import AsyncExecutor
+from threading import Thread
+
+
+class TestAsyncExecutor(unittest.TestCase):
+
+    def setUp(self):
+        self.a = AsyncExecutor()
+
+    @patch.object(Thread, 'start', Mock())
+    def test_run_async(self):
+        self.a.run_async(Mock(return_value=True))
+
+    def test_run(self):
+        self.a.run(Mock(side_effect=Exception()))

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
 
 
@@ -50,6 +49,8 @@ def requests_get(url, **kwargs):
     response = MockResponse()
     if url.startswith('http://local'):
         raise requests.exceptions.RequestException()
+    elif ':8011/patroni' in url:
+        response.content = '{"role": "replica", "xlog": {"replayed_location": 0}}'
     elif url.endswith('/members'):
         if url.startswith('http://error'):
             response.content = '[{}]'
@@ -92,6 +93,8 @@ def etcd_read(key, **kwargs):
         raise etcd.EtcdKeyNotFound
 
     response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [
+                {"key": "/service/batman5/failover", "value": "",
+                 "modifiedIndex": 1582, "createdIndex": 1582},
                 {"key": "/service/batman5/initialize", "value": "postgresql0",
                  "modifiedIndex": 1582, "createdIndex": 1582},
                 {"key": "/service/batman5/leader", "value": "postgresql1",
@@ -103,13 +106,13 @@ def etcd_read(key, **kwargs):
                  "modifiedIndex": 20437, "createdIndex": 20437},
                 {"key": "/service/batman5/members", "dir": True, "nodes": [
                     {"key": "/service/batman5/members/postgresql1",
-                     "value": "postgres://replicator:[email protected]:5434/postgres"
+                     "value": "postgres://replicator:[email protected]:5434/postgres" +
-                        + "?application_name=http://127.0.0.1:8009/patroni",
+                        "?application_name=http://127.0.0.1:8009/patroni",
                      "expiration": "2015-05-15T09:10:59.949384522Z", "ttl": 21,
                      "modifiedIndex": 20727, "createdIndex": 20727},
                     {"key": "/service/batman5/members/postgresql0",
-                     "value": "postgres://replicator:[email protected]:5433/postgres"
+                     "value": "postgres://replicator:[email protected]:5433/postgres" +
-                        + "?application_name=http://127.0.0.1:8008/patroni",
+                        "?application_name=http://127.0.0.1:8008/patroni",
                      "expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30,
                      "modifiedIndex": 20730, "createdIndex": 20730}],
                  "modifiedIndex": 1581, "createdIndex": 1581}], "modifiedIndex": 1581, "createdIndex": 1581}}
@@ -145,15 +148,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)
@@ -171,6 +165,11 @@ class TestClient(unittest.TestCase):
         self.client._base_uri = 'http://localhost:4001'
         self.client._machines_cache = ['http://localhost:2379']
         self.client.api_execute('/', 'GET')
+        self.client._update_machines_cache = False
+        self.client._base_uri = 'http://localhost:4001'
+        self.client._machines_cache = []
+        self.assertRaises(etcd.EtcdConnectionFailed, self.client.api_execute, '/', 'GET')
+        self.assertTrue(self.client._update_machines_cache)
 
     def test_get_srv_record(self):
         self.assertEquals(self.client.get_srv_record('blabla'), [])
@@ -199,7 +198,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):
 
@@ -242,8 +240,11 @@ class TestEtcd(unittest.TestCase):
         self.etcd._base_path = '/service/failed'
         self.assertFalse(self.etcd.attempt_to_acquire_leader())
 
+    def test_write_leader_optime(self):
+        self.etcd.write_leader_optime('0')
+
     def test_update_leader(self):
-        self.assertTrue(self.etcd.update_leader(MockPostgresql()))
+        self.assertTrue(self.etcd.update_leader())
 
     def test_initialize(self):
         self.assertFalse(self.etcd.initialize())
@@ -256,7 +257,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

@@ -1,11 +1,12 @@
+import etcd
 import unittest
 
-from mock import Mock, patch
+from mock import Mock, MagicMock, patch
-from patroni.dcs import Cluster, DCSError, Leader, Member
+from patroni.dcs import Cluster, Failover, Leader, Member
 from patroni.etcd import Client, Etcd
-from patroni.exceptions import PostgresException
+from patroni.exceptions import DCSError, PostgresException
 from patroni.ha import Ha
-from test_etcd import socket_getaddrinfo, etcd_read, etcd_write
+from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get
 
 
 def true(*args, **kwargs):
@@ -16,28 +17,33 @@ def false(*args, **kwargs):
     return False
 
 
-def get_cluster(initialize, leader):
+def get_cluster(initialize, leader, members, failover):
-    return Cluster(initialize, leader, None, None)
+    return Cluster(initialize, leader, None, members, failover)
 
 
 def get_cluster_not_initialized_without_leader():
-    return get_cluster(None, None)
+    return get_cluster(None, None, [], None)
 
 
-def get_cluster_initialized_without_leader():
+def get_cluster_initialized_without_leader(leader=False, failover=None):
-    return get_cluster(True, None)
+    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)
 
 
-def get_cluster_initialized_with_leader():
+def get_cluster_initialized_with_leader(failover=None):
-    return get_cluster(True, Leader(0, 0, 0,
+    return get_cluster_initialized_without_leader(leader=True, failover=failover)
-                       Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
-                              None, None, 28)))
 
 
 class MockPostgresql(Mock):
 
     name = 'postgresql0'
     role = 'replica'
+    state = 'running'
+    connection_string = 'postgres://foo@bar/postgres'
 
     def is_healthy(self):
         return True
@@ -51,6 +57,9 @@ class MockPostgresql(Mock):
     def is_leader(self):
         return True
 
+    def xlog_position(self):
+        return 0
+
     def last_operation(self):
         return 0
 
@@ -60,6 +69,25 @@ class MockPostgresql(Mock):
     def bootstrap(self, *args, **kwargs):
         return True
 
+    def check_replication_lag(self, last_leader_operation):
+        return True
+
+    def check_recovery_conf(self, leader):
+        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()
+
 
 class TestHa(unittest.TestCase):
 
@@ -71,23 +99,45 @@ 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.e.client.delete = Mock(side_effect=etcd.EtcdException())
-        self.ha.load_cluster_from_dcs()
+        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.ha.cluster = get_cluster_not_initialized_without_leader()
         self.ha.load_cluster_from_dcs = Mock()
 
-    def test_load_cluster_from_dcs(self):
+    def test_update_lock(self):
-        ha = Ha(self.p, self.e)
+        self.p.last_operation = Mock(side_effect=PostgresException(''))
-        ha.load_cluster_from_dcs()
+        self.assertTrue(self.ha.update_lock())
-        self.e.get_cluster = get_cluster_not_initialized_without_leader
-        ha.load_cluster_from_dcs()
 
-    def test_start_as_slave(self):
+    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')
 
+    def test_recover_replica_failed(self):
+        self.p.controldata = lambda: {'Database cluster state': 'in production'}
+        self.p.is_healthy = false
+        self.p.follow_the_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'failed to start postgres')
+
+    def test_recover_master_failed(self):
+        self.p.follow_the_leader = false
+        self.p.is_healthy = false
+        self.ha.has_lock = true
+        self.assertEquals(self.ha.run_cycle(), 'removed leader key after trying and failing to start postgres')
+
+    @patch('sys.exit', return_value=1)
+    @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True))
+    def test_sysid_no_match(self, exit_mock):
+        self.ha.run_cycle()
+        exit_mock.assert_called_once_with(1)
+
+    @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')
@@ -96,6 +146,7 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.run_cycle(), 'acquired session lock as a leader')
 
     def test_promoted_by_acquiring_lock(self):
+        self.ha.is_healthiest_node = true
         self.p.is_leader = false
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
 
@@ -104,16 +155,17 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.run_cycle(), 'demoted self due after trying and failing to obtain lock')
 
     def test_follow_new_leader_after_failing_to_obtain_lock(self):
+        self.ha.is_healthiest_node = true
         self.ha.acquire_lock = false
         self.p.is_leader = false
         self.assertEquals(self.ha.run_cycle(), 'following new leader after trying and failing to obtain lock')
 
     def test_demote_because_not_healthiest(self):
-        self.p.is_healthiest_node = false
+        self.ha.is_healthiest_node = false
         self.assertEquals(self.ha.run_cycle(), 'demoting self because i am not the healthiest node')
 
     def test_follow_new_leader_because_not_healthiest(self):
-        self.p.is_healthiest_node = false
+        self.ha.is_healthiest_node = false
         self.p.is_leader = false
         self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
 
@@ -148,13 +200,9 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader')
 
     def test_bootstrap_from_leader(self):
-        self.ha.cluster = get_cluster_initialized_with_leader()
-        self.assertEquals(self.ha.bootstrap(), 'bootstrapped from leader')
-
-    def test_bootstrap_from_leader_failed(self):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.p.bootstrap = false
-        self.assertEquals(self.ha.bootstrap(), 'failed to bootstrap from leader')
+        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()
@@ -174,3 +222,86 @@ class TestHa(unittest.TestCase):
         self.e.initialize = true
         self.p.bootstrap = Mock(side_effect=PostgresException("Could not bootstrap master PostgreSQL"))
         self.assertRaises(PostgresException, self.ha.bootstrap)
+
+    def test_reinitialize(self):
+        self.ha.schedule_reinitialize()
+        self.ha.schedule_reinitialize()
+        self.ha.run_cycle()
+        self.assertIsNone(self.ha._async_executor.scheduled_action)
+
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.ha.has_lock = true
+        self.ha.schedule_reinitialize()
+        self.ha.run_cycle()
+        self.assertIsNone(self.ha._async_executor.scheduled_action)
+
+        self.ha.has_lock = false
+        self.ha.schedule_reinitialize()
+        self.ha.run_cycle()
+
+    def test_restart(self):
+        self.assertEquals(self.ha.restart(), (True, 'restarted successfully'))
+        self.p.restart = false
+        self.assertEquals(self.ha.restart(), (False, 'restart failed'))
+        self.ha.schedule_reinitialize()
+        self.assertEquals(self.ha.restart(), (False, 'reinitialize already in progress'))
+
+    def test_restart_in_progress(self):
+        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.assertEquals(self.ha.run_cycle(), 'restart in progress')
+
+        self.ha.has_lock = true
+        self.assertEquals(self.ha.run_cycle(), 'updated leader lock during restart')
+
+        self.ha.update_lock = false
+        self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart')
+
+    @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.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.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.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.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.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.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.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')
+
+    def test__is_healthiest_node(self):
+        self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+        self.p.is_leader = false
+        self.ha.fetch_node_status = lambda e: (e, True, True, 0)  # accessible, in_recovery
+        self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+        self.ha.fetch_node_status = lambda e: (e, True, False, 0)  # accessible, not in_recovery
+        self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+        self.ha.fetch_node_status = lambda e: (e, True, True, 1)  # accessible, in_recovery, xlog location ahead
+        self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+        self.p.check_replication_lag = false
+        self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+
+    @patch('requests.get', requests_get)
+    def test_fetch_node_status(self):
+        member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'})
+        self.ha.fetch_node_status(member)
+        member = Member(0, 'test', 1, {'api_url': 'http://localhost:8011/patroni'})
+        self.ha.fetch_node_status(member)

tests/test_patroni.py

@@ -1,4 +1,3 @@
-import datetime
 import sys
 import time
 import unittest
@@ -6,7 +5,7 @@ import yaml
 
 from mock import Mock, patch
 from patroni.api import RestApiServer
-from patroni.dcs import Cluster, Member
+from patroni.async_executor import AsyncExecutor
 from patroni.etcd import Etcd
 from patroni import Patroni, main
 from patroni.zookeeper import ZooKeeper
@@ -26,6 +25,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(AsyncExecutor, 'run', Mock())
 class TestPatroni(unittest.TestCase):
 
     @patch.object(Client, 'machines')
@@ -48,7 +48,6 @@ class TestPatroni(unittest.TestCase):
         self.assertRaises(Exception, self.p.get_dcs, '', {})
 
     @patch('time.sleep', Mock(side_effect=SleepException()))
-    @patch.object(Patroni, 'initialize', Mock())
     @patch.object(Etcd, 'delete_leader', Mock())
     @patch.object(Client, 'machines')
     def test_patroni_main(self, mock_machines):
@@ -56,7 +55,6 @@ 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())):
@@ -64,8 +62,6 @@ class TestPatroni(unittest.TestCase):
 
     @patch('time.sleep', Mock(side_effect=SleepException()))
     def test_run(self):
-        self.p.touch_member = self.touch_member
-        self.p.ha.state_handler.sync_replication_slots = time_sleep
         self.p.ha.dcs.watch = time_sleep
         self.assertRaises(SleepException, self.p.run)
 
@@ -73,24 +69,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])
-        self.p.touch_member()
-
-    def test_patroni_initialize(self):
-        self.p.touch_member = self.touch_member
-        self.p.initialize()
-
     def test_schedule_next_run(self):
         self.p.ha.dcs.watch = Mock(return_value=True)
         self.p.schedule_next_run()

tests/test_postgresql.py

@@ -1,14 +1,22 @@
+import mock  # for the mock.call method, importing it without a namespace breaks python3
 import os
 import psycopg2
 import shutil
 import unittest
 
-from mock import Mock, patch
+from six.moves import builtins
+from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 from patroni.dcs import Cluster, Leader, Member
 from patroni.exceptions import PostgresException, PostgresConnectionException
 from patroni.postgresql import Postgresql
 from patroni.utils import RetryFailedError
 from test_ha import false
+import subprocess
+
+
+def is_file_raise_on_backup(*args, **kwargs):
+    if args[0].endswith('.backup'):
+        raise Exception("foo")
 
 
 class MockCursor:
@@ -25,18 +33,9 @@ class MockCursor:
             raise RetryFailedError('retry')
         elif sql.startswith('SELECT slot_name'):
             self.results = [('blabla',), ('foobar',)]
-        elif sql.startswith('SELECT pg_current_xlog_location()'):
-            self.results = [(0,)]
-        elif sql.startswith('SELECT pg_is_in_recovery(), %s'):
-            if params[0][0] == 1:
-                raise psycopg2.OperationalError()
-            elif params[0][0] == 2:
-                self.results = [(True, -1)]
-            else:
-                self.results = [(False, 0)]
         elif sql.startswith('SELECT pg_xlog_location_diff'):
             self.results = [(0,)]
-        elif sql.startswith('SELECT pg_is_in_recovery()'):
+        elif sql == 'SELECT pg_is_in_recovery()':
             self.results = [(False, )]
         elif sql.startswith('SELECT to_char(pg_postmaster_start_time'):
             self.results = [('', True, '', '', '', False)]
@@ -79,22 +78,93 @@ class MockConnect(Mock):
     def cursor(self):
         return MockCursor(self)
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        pass
+
+
+def pg_controldata_string(*args, **kwargs):
+    return b"""
+pg_control version number:            942
+Catalog version number:               201509161
+Database system identifier:           6200971513092291716
+Database cluster state:               shut down in recovery
+pg_control last modified:             Fri Oct  2 10:57:06 2015
+Latest checkpoint location:           0/30000C8
+Prior checkpoint location:            0/2000060
+Latest checkpoint's REDO location:    0/3000090
+Latest checkpoint's REDO WAL file:    000000020000000000000003
+Latest checkpoint's TimeLineID:       2
+Latest checkpoint's PrevTimeLineID:   2
+Latest checkpoint's full_page_writes: on
+Latest checkpoint's NextXID:          0/943
+Latest checkpoint's NextOID:          24576
+Latest checkpoint's NextMultiXactId:  1
+Latest checkpoint's NextMultiOffset:  0
+Latest checkpoint's oldestXID:        931
+Latest checkpoint's oldestXID's DB:   1
+Latest checkpoint's oldestActiveXID:  943
+Latest checkpoint's oldestMultiXid:   1
+Latest checkpoint's oldestMulti's DB: 1
+Latest checkpoint's oldestCommitTs:   0
+Latest checkpoint's newestCommitTs:   0
+Time of latest checkpoint:            Fri Oct  2 10:56:54 2015
+Fake LSN counter for unlogged rels:   0/1
+Minimum recovery ending location:     0/30241F8
+Min recovery ending loc's timeline:   2
+Backup start location:                0/0
+Backup end location:                  0/0
+End-of-backup record required:        no
+wal_level setting:                    hot_standby
+Current wal_log_hints setting:                on
+Current max_connections setting:              100
+Current max_worker_processes setting:         8
+Current max_prepared_xacts setting:           0
+Current max_locks_per_xact setting:           64
+Current track_commit_timestamp setting:       off
+Maximum data alignment:               8
+Database block size:                  8192
+Blocks per segment of large relation: 131072
+WAL block size:                       8192
+Bytes per WAL segment:                16777216
+Maximum length of identifiers:        64
+Maximum columns in an index:          32
+Maximum size of a TOAST chunk:        1996
+Size of a large-object chunk:         2048
+Date/time type storage:               64-bit integers
+Float4 argument passing:              by value
+Float8 argument passing:              by value
+Data page checksum version:           0
+"""
+
+
+def postmaster_opts_string(*args, **kwargs):
+    return '/usr/local/pgsql/bin/postgres "-D" "data/postgresql0" "--listen_addresses=127.0.0.1" \
+"--port=5432" "--hot_standby=on" "--wal_keep_segments=8" "--wal_level=hot_standby" \
+"--archive_command=mkdir -p ../wal_archive && cp %p ../wal_archive/%f" "--wal_log_hints=on" \
+"--max_wal_senders=5" "--archive_timeout=1800s" "--archive_mode=on" "--max_replication_slots=5"\n'
+
 
 def psycopg2_connect(*args, **kwargs):
     return MockConnect()
 
 
 @patch('subprocess.call', Mock(return_value=0))
-@patch('shutil.copy', Mock())
 @patch('psycopg2.connect', psycopg2_connect)
+@patch('shutil.copy', Mock())
 class TestPostgresql(unittest.TestCase):
 
+    @patch('subprocess.call', Mock(return_value=0))
+    @patch('psycopg2.connect', psycopg2_connect)
     def setUp(self):
         self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': 'data/test0',
                              'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432',
                              'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'],
                              'superuser': {'password': ''},
                              'admin': {'username': 'admin', 'password': 'admin'},
+                             'pg_rewind': {'username': 'admin', 'password': 'admin'},
                              'replication': {'username': 'replicator',
                                              'password': 'rep-pass',
                                              'network': '127.0.0.1/32'},
@@ -106,10 +176,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.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
-        self.leader = Leader(-1, None, 28, self.leadermem)
+        self.leader = Leader(-1, 28, self.leadermem)
-        self.other = Member(0, 'test1', 'postgres://replicator:[email protected]:5433/postgres', None, None, 28)
+        self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres'})
-        self.me = Member(0, 'test0', 'postgres://replicator:[email protected]:5434/postgres', None, None, 28)
+        self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:[email protected]:5434/postgres'})
 
     def tearDown(self):
         shutil.rmtree('data')
@@ -121,23 +191,77 @@ class TestPostgresql(unittest.TestCase):
         self.assertTrue(self.p.initialize())
         self.assertTrue(os.path.exists(os.path.join(self.p.data_dir, 'pg_hba.conf')))
 
-    def test_start_stop(self):
+    def test_start(self):
-        self.assertFalse(self.p.start())
+        self.assertTrue(self.p.start())
         self.p.is_running = false
-        with open(os.path.join(self.p.data_dir, 'postmaster.pid'), 'w'):
+        open(os.path.join(self.p.data_dir, 'postmaster.pid'), 'w').close()
-            pass
         self.assertTrue(self.p.start())
-        self.assertTrue(self.p.stop())
 
+    def test_stop(self):
+        self.assertTrue(self.p.stop())
+        with patch('subprocess.call', Mock(return_value=1)):
+            self.assertTrue(self.p.stop())
+            self.p.is_running = Mock(return_value=True)
+            self.assertFalse(self.p.stop())
+
+    def test_restart(self):
+        self.p.start = false
+        self.p.is_running = false
+        self.assertFalse(self.p.restart())
+        self.assertEquals(self.p.state, 'restart failed (restarting)')
+
+    @patch.object(builtins, 'open', MagicMock())
+    def test_write_pgpass(self):
+        self.p.write_pgpass({'host': 'localhost', 'port': '5432', 'user': 'foo', 'password': 'bar'})
+
+    @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
     def test_sync_from_leader(self):
         self.assertTrue(self.p.sync_from_leader(self.leader))
 
-    def test_follow_the_leader(self):
+    @patch('subprocess.call', side_effect=Exception("Test"))
-        self.p.demote(self.leader)
+    @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
+    def test_pg_rewind(self, mock_call):
+        self.assertTrue(self.p.rewind(self.leader))
+        subprocess.call = mock_call
+        self.assertFalse(self.p.rewind(self.leader))
+
+    @patch('patroni.postgresql.Postgresql.rewind', return_value=False)
+    @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True))
+    @patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1))
+    @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
+    def test_follow_the_leader(self, mock_pg_rewind):
+        self.p.demote()
         self.p.follow_the_leader(None)
-        self.p.demote(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))
+        self.p.rewind = mock_pg_rewind
+        self.p.follow_the_leader(self.leader)
+        self.p.require_rewind()
+        with mock.patch('os.path.islink', MagicMock(return_value=True)):
+            with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
+                with mock.patch('os.unlink', MagicMock(return_value=True)):
+                    self.p.follow_the_leader(self.leader, recovery=True)
+        self.p.require_rewind()
+        with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)):
+            self.p.rewind.return_value = True
+            self.p.follow_the_leader(self.leader, recovery=True)
+            self.p.rewind.return_value = False
+            self.p.follow_the_leader(self.leader, recovery=True)
+
+    def test_can_rewind(self):
+        tmp = self.p.pg_rewind
+        self.p.pg_rewind = None
+        self.assertFalse(self.p.can_rewind)
+        self.p.pg_rewind = tmp
+        with mock.patch('subprocess.call', MagicMock(return_value=1)):
+            self.assertFalse(self.p.can_rewind)
+        with mock.patch('subprocess.call', side_effect=OSError("foo")):
+            self.assertFalse(self.p.can_rewind)
+        tmp = self.p.controldata()
+        self.p.controldata = lambda: {'wal_log_hints setting': 'on'}
+        self.assertTrue(self.p.can_rewind)
+        self.p.controldata = tmp
 
     def test_create_replica(self):
         self.p.delete_trigger_file = Mock(side_effect=OSError())
@@ -151,29 +275,25 @@ class TestPostgresql(unittest.TestCase):
 
     def test_sync_replication_slots(self):
         self.p.start()
-        cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem])
+        cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem], None)
+        self.p.sync_replication_slots(cluster)
+        self.p.query = Mock(side_effect=psycopg2.OperationalError)
+        self.p.schedule_load_slots = True
         self.p.sync_replication_slots(cluster)
 
     @patch.object(MockConnect, 'closed', 2)
     def test__query(self):
         self.assertRaises(PostgresConnectionException, self.p._query, 'blabla')
+        self.p._state = 'restarting'
+        self.assertRaises(RetryFailedError, self.p._query, 'blabla')
 
     def test_query(self):
         self.p.query('select 1')
         self.assertRaises(PostgresConnectionException, self.p.query, 'RetryFailedError')
         self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
 
-    def test_is_healthiest_node(self):
+    def test_is_leader(self):
-        cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem])
+        self.assertTrue(self.p.is_leader())
-        self.assertTrue(self.p.is_healthiest_node(cluster))
-        self.p.is_leader = false
-        self.assertFalse(self.p.is_healthiest_node(cluster))
-        self.p.xlog_position = lambda: 1
-        self.assertTrue(self.p.is_healthiest_node(cluster))
-        self.p.xlog_position = lambda: 2
-        self.assertFalse(self.p.is_healthiest_node(cluster))
-        self.p.config['maximum_lag_on_failover'] = -3
-        self.assertFalse(self.p.is_healthiest_node(cluster))
 
     def test_reload(self):
         self.assertTrue(self.p.reload())
@@ -184,6 +304,7 @@ class TestPostgresql(unittest.TestCase):
         self.assertFalse(self.p.is_healthy())
 
     def test_promote(self):
+        self.p._role = 'replica'
         self.assertTrue(self.p.promote())
         self.assertTrue(self.p.promote())
 
@@ -202,6 +323,9 @@ class TestPostgresql(unittest.TestCase):
         self.p.query = Mock(side_effect=psycopg2.OperationalError("not supported"))
         self.assertTrue(self.p.stop())
 
+    def test_check_replication_lag(self):
+        self.assertTrue(self.p.check_replication_lag(0))
+
     @patch('os.rename', Mock())
     @patch('os.path.isdir', Mock(return_value=True))
     def test_move_data_directory(self):
@@ -210,10 +334,12 @@ class TestPostgresql(unittest.TestCase):
         with patch('os.rename', Mock(side_effect=OSError())):
             self.p.move_data_directory()
 
+    @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
     def test_bootstrap(self):
+        with patch('subprocess.call', Mock(return_value=1)):
             self.assertRaises(PostgresException, self.p.bootstrap)
-        self.p.start = Mock(return_value=True)
         self.p.bootstrap()
+        self.p.bootstrap(self.leader)
 
     def test_remove_data_directory(self):
         self.p.data_dir = 'data_dir'
@@ -226,3 +352,104 @@ class TestPostgresql(unittest.TestCase):
         with patch('os.unlink', Mock(side_effect=Exception)):
             self.p.remove_data_directory()
         self.p.remove_data_directory()
+
+    @patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string))
+    @patch('subprocess.check_output', side_effect=subprocess.CalledProcessError)
+    @patch('subprocess.check_output', side_effect=Exception('Failed'))
+    def test_controldata(self, check_output_call_error, check_output_generic_exception):
+        data = self.p.controldata()
+        self.assertEquals(len(data), 50)
+        self.assertEquals(data['Database cluster state'], 'shut down in recovery')
+        self.assertEquals(data['wal_log_hints setting'], 'on')
+        self.assertEquals(int(data['Database block size']), 8192)
+
+        subprocess.check_output = check_output_call_error
+        data = self.p.controldata()
+        self.assertEquals(data, dict())
+
+        subprocess.check_output = check_output_generic_exception
+        self.assertRaises(Exception, self.p.controldata())
+
+    def test_read_postmaster_opts(self):
+        m = mock_open(read_data=postmaster_opts_string())
+        with patch.object(builtins, 'open', m):
+            data = self.p.read_postmaster_opts()
+            self.assertEquals(data['wal_level'], 'hot_standby')
+            self.assertEquals(int(data['max_replication_slots']), 5)
+            self.assertEqual(data.get('D'), None)
+
+            m.side_effect = IOError("foo")
+            data = self.p.read_postmaster_opts()
+            self.assertEqual(data, dict())
+
+            m.side_effect = Exception("foo")
+            self.assertRaises(Exception, self.p.read_postmaster_opts())
+
+    @patch('subprocess.Popen')
+    @patch.object(builtins, 'open', MagicMock(return_value=42))
+    def test_single_user_mode(self, subprocess_popen_mock):
+        subprocess_popen_mock.return_value.wait.return_value = 0
+        self.assertEquals(self.p.single_user_mode(options=dict(archive_mode='on', archive_command='false')), 0)
+        subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir,
+                                                      '-c', 'archive_command=false', '-c', 'archive_mode=on',
+                                                       'postgres'], stdin=subprocess.PIPE,
+                                                      stdout=42,
+                                                      stderr=subprocess.STDOUT)
+        subprocess_popen_mock.reset_mock()
+        self.assertEquals(self.p.single_user_mode(command="CHECKPOINT"), 0)
+        subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir,
+                                                      'postgres'], stdin=subprocess.PIPE,
+                                                      stdout=42,
+                                                      stderr=subprocess.STDOUT)
+        subprocess_popen_mock.return_value = None
+        self.assertEquals(self.p.single_user_mode(), 1)
+
+    def fake_listdir(path):
+        if path.endswith(os.path.join('pg_xlog', 'archive_status')):
+            return ["a", "b", "c"]
+        return []
+
+    @patch('os.listdir', MagicMock(side_effect=fake_listdir))
+    @patch('os.path.isdir', MagicMock(return_value=True))
+    @patch('os.unlink', return_value=True)
+    @patch('os.remove', return_value=True)
+    @patch('os.path.islink', return_value=False)
+    @patch('os.path.isfile', return_value=True)
+    def test_cleanup_archive_status(self, mock_file, mock_link, mock_remove, mock_unlink):
+        ap = os.path.join(self.p.data_dir, 'pg_xlog', 'archive_status/')
+        self.p.cleanup_archive_status()
+        mock_remove.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')])
+        mock_unlink.assert_not_called()
+
+        mock_remove.reset_mock()
+
+        mock_file.return_value = False
+        mock_link.return_value = True
+        self.p.cleanup_archive_status()
+        mock_unlink.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')])
+        mock_remove.assert_not_called()
+
+        mock_unlink.reset_mock()
+        mock_remove.reset_mock()
+
+        mock_file.side_effect = Exception("foo")
+        mock_link.side_effect = Exception("foo")
+        self.p.cleanup_archive_status()
+        mock_unlink.assert_not_called()
+        mock_remove.assert_not_called()
+
+    @patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string))
+    def test_sysid(self):
+        self.assertEqual(self.p.sysid, "6200971513092291716")
+
+    @patch('os.path.isfile', MagicMock(return_value=True))
+    @patch('shutil.copy', side_effect=Exception)
+    def test_save_configuration_files(self, mock_copy):
+        shutil.copy = mock_copy
+        self.p.save_configuration_files()
+
+    @patch('os.path.isfile', MagicMock(side_effect=is_file_raise_on_backup))
+    @patch('shutil.copy', side_effect=Exception)
+    def test_restore_configuration_files(self, mock_copy):
+        shutil.copy = mock_copy
+        self.p.restore_configuration_files()

tests/test_utils.py

@@ -67,7 +67,9 @@ class TestRetrySleeper(unittest.TestCase):
         self.assertRaises(RetryFailedError, retry, self._fail(times=100))
 
     def test_copy(self):
-        _sleep = lambda t: None
+        def _sleep(t):
+            None
+
         retry = self._makeOne(sleep_func=_sleep)
         rcopy = retry.copy()
         self.assertTrue(rcopy.sleep_func is _sleep)

tests/test_zookeeper.py

@@ -7,7 +7,7 @@ from patroni.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperErr
 from kazoo.client import KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.protocol.states import ZnodeStat
-from test_etcd import MockPostgresql, SleepException, requests_get
+from test_etcd import SleepException, requests_get
 
 
 class MockKazooClient(Mock):
@@ -31,7 +31,7 @@ class MockKazooClient(Mock):
         elif '/members/' in path:
             return (
                 b'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni',
-                ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
+                ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0 if self.exists else -1, 0, 0, 0)
             )
         elif path.endswith('/optime/leader'):
             return (b'1', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
@@ -41,14 +41,15 @@ class MockKazooClient(Mock):
             return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
         elif path.endswith('/initialize'):
             return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
+        return (b'', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
 
     def get_children(self, path, watch=None, include_data=False):
         if not isinstance(path, six.string_types):
             raise TypeError("Invalid type for 'path' (string expected)")
-        if path == '/no_node':
+        if path.startswith('/no_node'):
             raise NoNodeError
         elif path in ['/service/bla/', '/service/test/']:
-            return ['initialize', 'leader', 'members', 'optime']
+            return ['initialize', 'leader', 'members', 'optime', 'failover']
         return ['foo', 'bar', 'buzz']
 
     def create(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False):
@@ -68,6 +69,14 @@ 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
+            elif value == b'ok':
+                return
         raise NoNodeError
 
     def delete(self, path, version=-1, recursive=False):
@@ -79,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
 
 
@@ -110,6 +121,8 @@ class TestZooKeeper(unittest.TestCase):
     def test__inner_load_cluster(self):
         self.zk._base_path = self.zk._base_path.replace('test', 'bla')
         self.zk._inner_load_cluster()
+        self.zk._base_path = self.zk._base_path = '/no_node'
+        self.zk._inner_load_cluster()
 
     def test_get_cluster(self):
         self.assertRaises(ZooKeeperError, self.zk.get_cluster)
@@ -119,6 +132,11 @@ class TestZooKeeper(unittest.TestCase):
         self.zk.touch_member('foo')
         self.zk.delete_leader()
 
+    def test_set_failover_value(self):
+        self.zk.set_failover_value('')
+        self.zk.set_failover_value('ok')
+        self.zk.set_failover_value('Exception')
+
     def test_initialize(self):
         self.assertFalse(self.zk.initialize())
 
@@ -126,21 +144,33 @@ 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.fetch_cluster = True
+        self.zk.get_cluster()
         self.zk.touch_member('retry')
 
     def test_take_leader(self):
         self.zk.take_leader()
 
     def test_update_leader(self):
-        self.zk.last_leader_operation = -1
+        self.assertTrue(self.zk.update_leader())
-        self.assertTrue(self.zk.update_leader(MockPostgresql()))
+
+    def test_write_leader_optime(self):
+        self.zk.last_leader_operation = '0'
+        self.zk.write_leader_optime('1')
         self.zk._base_path = self.zk._base_path.replace('test', 'bla')
-        self.zk.last_leader_operation = -1
+        self.zk.write_leader_optime('2')
-        self.assertTrue(self.zk.update_leader(MockPostgresql()))
 
     def test_watch(self):
         self.zk.watch(0)
-        self.zk.cluster_event.isSet = lambda: False
+        self.zk.event.isSet = lambda: True
         self.zk.watch(0)