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