From c218054d05e4ddbaee5a144ae439e7c6ffe72bdb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 28 Sep 2015 17:00:42 +0200 Subject: [PATCH] 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 | 11 +-- patroni/api.py | 9 +-- patroni/dcs.py | 25 ++++++- patroni/etcd.py | 22 ++++-- patroni/ha.py | 147 +++++++++++++++++++++++++++++++++++++-- patroni/postgresql.py | 32 +-------- patroni/zookeeper.py | 35 +++++++--- tests/test_etcd.py | 4 ++ tests/test_ha.py | 88 +++++++++++++++++++---- tests/test_patroni.py | 7 +- tests/test_postgresql.py | 29 ++------ tests/test_zookeeper.py | 19 ++++- 12 files changed, 320 insertions(+), 108 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index cadf335a..6ca96df0 100644 --- a/patroni/__init__.py +++ b/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: diff --git a/patroni/api.py b/patroni/api.py index ebb732a4..d507bfb3 100644 --- a/patroni/api.py +++ b/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, diff --git a/patroni/dcs.py b/patroni/dcs.py index ebf625d6..fb034322 100644 --- a/patroni/dcs.py +++ b/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() diff --git a/patroni/etcd.py b/patroni/etcd.py index b189c188..c2995da9 100644 --- a/patroni/etcd.py +++ b/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 diff --git a/patroni/ha.py b/patroni/ha.py index 64179a78..628b9c41 100644 --- a/patroni/ha.py +++ b/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): diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 2bd372cf..bc3188ed 100644 --- a/patroni/postgresql.py +++ b/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: diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 5d221ac6..7458ae46 100644 --- a/patroni/zookeeper.py +++ b/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 diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 18e8f22e..f55aa15e 100644 --- a/tests/test_etcd.py +++ b/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", diff --git a/tests/test_ha.py b/tests/test_ha.py index 70018672..6e50c629 100644 --- a/tests/test_ha.py +++ b/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:rep-pass@127.0.0.1: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:rep-pass@127.0.0.1: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:rep-pass@127.0.0.1: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) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 7cac2eba..8936d18b 100644 --- a/tests/test_patroni.py +++ b/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() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index d1ab7e8d..9fe012a7 100644 --- a/tests/test_postgresql.py +++ b/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): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index b04f5b5a..039faa58 100644 --- a/tests/test_zookeeper.py +++ b/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()