Implement manual failover

Implementation is done on top of feature/is-healthiest-via-api and
feature/api branches.
In order to trigger manual failover one has to create 'failover' key in
a configuration store with the value in following format:
'leader_name:member_name'
leader_name can be empty or should match with the name of current leader
member_name can be empty or should match with the name one of cluster
nodes
Leader always checks that either desired member (if specified) or one of
the memners is accessible and healthy before demote.
After leader has deomted himself other nodes are performig checks that
desired node is healthy. If it is not they are participating in a leader
race. In some cases (when accidently there is no healthy nodes) former
leader can also participate in a leader race.

Current implementation does not provide REST API endpoint for a manual
failover.

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__)
@@ -42,12 +42,6 @@ class Patroni:
                     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)
-
     def schedule_next_run(self):
         self.next_run += self.nap_time
         current_time = time.time()
@@ -62,12 +56,12 @@ 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')
+            self.touch_member()
             reap_children()
             self.schedule_next_run()
 
@@ -85,7 +79,6 @@ def main():
         config = yaml.load(f)
 
     patroni = Patroni(config)
-    patroni.initialize()
     try:
         patroni.run()
     except KeyboardInterrupt:

patroni/api.py

@@ -143,10 +143,11 @@ 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
-                                            ELSE pg_current_xlog_location() END,
-                                       pg_last_xlog_receive_location(),
-                                       pg_last_xlog_replay_location(),
+                                            THEN 0
+                                            ELSE pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint
+                                       END,
+                                       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 {
                 'state': self.server.patroni.postgresql.state,

patroni/dcs.py

@@ -56,7 +56,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 +72,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 +85,7 @@ class AbstractDCS:
 
     _INITIALIZE = 'initialize'
     _LEADER = 'leader'
+    _FAILOVER = 'failover'
     _MEMBERS = 'members/'
     _OPTIME = 'optime'
     _LEADER_OPTIME = _OPTIME + '/' + _LEADER
@@ -109,6 +119,10 @@ 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)
@@ -143,6 +157,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()

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, parse_connection_string
+from patroni.exceptions import DCSError
 from patroni.utils import Retry, RetryFailedError, sleep
 from requests.exceptions import RequestException
 
@@ -80,7 +81,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'])
@@ -195,9 +196,14 @@ class Etcd(AbstractDCS):
                 member = ([m for m in members if m.name == leader.value] or [member])[0]
                 leader = Leader(leader.modifiedIndex, leader.expiration, 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')
@@ -221,6 +227,10 @@ class Etcd(AbstractDCS):
             pass
         return False
 
+    @catch_etcd_errors
+    def set_failover_value(self, value, index=None):
+        return self.client.write(self.failover_path, value, prevIndex=index or 0)
+
     @catch_etcd_errors
     def write_leader_optime(self, last_operation):
         return self.client.set(self.leader_optime_path, last_operation)
@@ -231,7 +241,7 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def initialize(self):
-        return self.client.write(self.initialize_path, self._name, prevExist=False)
+        return self.retry(self.client.write, self.initialize_path, self._name, prevExist=False)
 
     @catch_etcd_errors
     def delete_leader(self):
@@ -239,7 +249,7 @@ 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):
         # watch on leader key changes if it is defined and current node is not lock owner

patroni/ha.py

@@ -1,7 +1,9 @@
 import logging
 import psycopg2
+import requests
 
 from patroni.exceptions import DCSError, PostgresConnectionException
+from multiprocessing.pool import ThreadPool
 from threading import Lock
 
 logger = logging.getLogger(__name__)
@@ -9,9 +11,9 @@ logger = logging.getLogger(__name__)
 
 class Ha:
 
-    def __init__(self, state_handler, etcd):
+    def __init__(self, state_handler, dcs):
         self.state_handler = state_handler
-        self.dcs = etcd
+        self.dcs = dcs
         self.cluster = None
         self.old_cluster = None
         self.scheduled_action = None
@@ -100,9 +102,141 @@ 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.old_cluster.members + self.cluster.members}
+        return self._is_healthiest_node(members.values())
+
+    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.state_handler.follow_the_leader(None)
+                    self.cluster = None
+                    if self.dcs.delete_leader():
+                        return 'manual failover: demoted self and released leader lock'
+                    else:
+                        return 'manual failover: demoted self but failed to release leader lock'
+                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:
@@ -114,6 +248,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')
@@ -220,7 +359,7 @@ class Ha:
                 return self.process_healthy_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)
                 return 'demoted self because DCS is not accessible and i was a leader'
         except (psycopg2.Error, PostgresConnectionException):

patroni/postgresql.py

@@ -266,36 +266,8 @@ class Postgresql:
             return False
         return True
 
-    def is_healthiest_node(self, cluster):
-        if self.is_leader():
-            return True
-
-        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 check_replication_lag(self, last_leader_operation):
+        return last_leader_operation - self.xlog_position() <= self.config.get('maximum_lag_on_failover', 0)
 
     def write_pg_hba(self):
         with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f:

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, parse_connection_string
+from patroni.exceptions import DCSError
 from patroni.utils import sleep
 from requests.exceptions import RequestException
 
@@ -134,7 +135,7 @@ class ZooKeeper(AbstractDCS):
 
     def _inner_load_cluster(self):
         self.cluster_event.clear()
-        nodes = set(self.get_children(self.client_path('')))
+        nodes = set(self.get_children(self.client_path(''), self.cluster_watcher))
 
         # get initialize flag
         initialize = self._INITIALIZE in nodes
@@ -143,7 +144,7 @@ class ZooKeeper(AbstractDCS):
         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:
@@ -157,10 +158,15 @@ class ZooKeeper(AbstractDCS):
                 leader = Leader(leader[1].version, None, None, member)
                 self.fetch_cluster = member.index == -1
 
+        # failover key
+        failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
+        if failover:
+            failover = Failover.from_node(failover[1].version, failover[0])
+
         # get last leader operation
         self.last_leader_operation = self.get_node(self.leader_optime_path) if self.fetch_cluster else None
         self.last_leader_operation = 0 if self.last_leader_operation is None else int(self.last_leader_operation[0])
-        self.cluster = Cluster(initialize, leader, self.last_leader_operation, members)
+        self.cluster = Cluster(initialize, leader, self.last_leader_operation, members, failover)
 
     def get_cluster(self):
         if self.exhibitor and self.exhibitor.poll():
@@ -188,11 +194,21 @@ class ZooKeeper(AbstractDCS):
         ret or logger.info('Could not take out TTL lock')
         return ret
 
+    def set_failover_value(self, value, index=None):
+        try:
+            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.encode('utf-8')))
+        except:
+            logging.exception('foo')
+            return False
+
     def initialize(self):
         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):
+        if not self.fetch_cluster and 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')
@@ -201,6 +217,9 @@ class ZooKeeper(AbstractDCS):
             return True
         except NodeExistsError:
             try:
+                node = self.get_node(path)
+                if node and self.client.client_id is not None and node[1].ephemeralOwner == self.client.client_id[0]:
+                    return True
                 self.client.retry(self.client.delete, path)
                 self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True)
                 return True
@@ -230,8 +249,8 @@ class ZooKeeper(AbstractDCS):
         return True
 
     def delete_leader(self):
-        if isinstance(self.cluster, Cluster) and self.cluster.leader.name == self._name:
-            self.client.delete(self.leader_path, version=self.cluster.leader.index)
+        self.client.restart()
+        return True
 
     def _cancel_initialization(self):
         node = self.get_node(self.initialize_path)
@@ -248,5 +267,5 @@ class ZooKeeper(AbstractDCS):
         self.cluster_event.wait(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 True
         return False

tests/test_etcd.py

@@ -50,6 +50,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 +94,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",

tests/test_ha.py

@@ -1,11 +1,11 @@
 import unittest
 
 from mock import Mock, 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,22 +16,25 @@ def false(*args, **kwargs):
     return False
 
 
-def get_cluster(initialize, leader):
-    return Cluster(initialize, leader, None, None)
+def get_cluster(initialize, leader, members, failover):
+    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():
-    return get_cluster(True, None)
+def get_cluster_initialized_without_leader(leader=False, failover=None):
+    m = Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
+               'http://127.0.0.1:8008/patroni', None, 28)
+    l = Leader(0, 0, 0, m) if leader else None
+    o = Member(0, 'other', 'postgres://replicator:[email protected]:5436/postgres',
+               'http://127.0.0.1:8011/patroni', None, 28)
+    return get_cluster(True, l, [m, o], failover)
 
 
-def get_cluster_initialized_with_leader():
-    return get_cluster(True, Leader(0, 0, 0,
-                       Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
-                              None, None, 28)))
+def get_cluster_initialized_with_leader(failover=None):
+    return get_cluster_initialized_without_leader(leader=True, failover=failover)
 
 
 class MockPostgresql(Mock):
@@ -51,6 +54,9 @@ class MockPostgresql(Mock):
     def is_leader(self):
         return True
 
+    def xlog_position(self):
+        return 0
+
     def last_operation(self):
         return 0
 
@@ -60,6 +66,9 @@ class MockPostgresql(Mock):
     def bootstrap(self, *args, **kwargs):
         return True
 
+    def check_replication_lag(self, last_leader_operation):
+        return True
+
 
 class TestHa(unittest.TestCase):
 
@@ -111,6 +120,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')
 
@@ -119,16 +129,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')
 
@@ -221,3 +232,52 @@ class TestHa(unittest.TestCase):
 
         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: demoted self but failed to release leader lock')
+        self.ha.cluster = get_cluster_initialized_with_leader(f)
+        self.e.client.delete = Mock(return_value=True)
+        self.assertEquals(self.ha.run_cycle(), 'manual failover: demoted self and released leader lock')
+
+    @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', '', 'http://127.0.0.1:8011/patroni', None, None)
+        self.ha.fetch_node_status(member)
+        member = Member(0, 'test', '', 'http://localhost:8011/patroni', None, None)
+        self.ha.fetch_node_status(member)

tests/test_patroni.py

@@ -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):
@@ -84,13 +83,9 @@ class TestPatroni(unittest.TestCase):
         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.ha.cluster = Cluster(True, member, 0, [member], None)
         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

@@ -25,18 +25,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)]
@@ -164,7 +155,7 @@ 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)
 
     @patch.object(MockConnect, 'closed', 2)
@@ -178,17 +169,8 @@ class TestPostgresql(unittest.TestCase):
         self.assertRaises(PostgresConnectionException, self.p.query, 'RetryFailedError')
         self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
 
-    def test_is_healthiest_node(self):
-        cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem])
-        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_is_leader(self):
+        self.assertTrue(self.p.is_leader())
 
     def test_reload(self):
         self.assertTrue(self.p.reload())
@@ -218,6 +200,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):

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,6 +41,7 @@ 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):
@@ -48,7 +49,7 @@ class MockKazooClient(Mock):
         if path == '/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,11 @@ 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/failover':
+            if value == b'Exception':
+                raise Exception
+            elif value == b'ok':
+                return
         raise NoNodeError
 
     def delete(self, path, version=-1, recursive=False):
@@ -119,6 +125,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())
 
@@ -129,6 +140,8 @@ class TestZooKeeper(unittest.TestCase):
         self.zk.touch_member('new')
         self.zk.touch_member('exists')
         self.zk.touch_member('retry')
+        self.zk.client.exists = True
+        self.zk.touch_member('retry')
 
     def test_take_leader(self):
         self.zk.take_leader()