From 1290b30b847d9b8240a75bba710f12c8bbbab1eb Mon Sep 17 00:00:00 2001 From: Ants Aasma Date: Thu, 8 Dec 2016 15:44:27 +0200 Subject: [PATCH] Introduce starting state and master start timeout. (#295) Previously pg_ctl waited for a timeout and then happily trodded on considering PostgreSQL to be running. This caused PostgreSQL to show up in listings as running when it was actually not and caused a race condition that resulted in either a failover or a crash recovery or a crash recovery interrupted by failover and a missed rewind. This change adds a master_start_timeout parameter and introduces a new state for the main run_cycle loop: starting. When master_start_timeout is zero we will fail over as soon as there is a failover candidate. Otherwise PostgreSQL will be started, but once master_start_timeout expires we will stop and release leader lock if failover is possible. Once failover succeeds or fails (no leader and no one to take the role) we continue with normal processing. While we are waiting for the master timeout we handle manual failover requests. * Introduce timeout parameter to restart. When restart timeout is set master becomes eligible for failover after that timeout expires regardless of master_start_time. Immediate restart calls will wait for this timeout to pass, even when node is a standby. --- docs/SETTINGS.rst | 1 + patroni/__init__.py | 25 +++- patroni/api.py | 12 +- patroni/config.py | 1 + patroni/ctl.py | 7 +- patroni/ha.py | 248 ++++++++++++++++++++++++++------- patroni/postgresql.py | 292 ++++++++++++++++++++++++++++++++------- postgres0.yml | 3 +- tests/test_api.py | 8 ++ tests/test_config.py | 5 +- tests/test_ctl.py | 3 + tests/test_etcd.py | 2 +- tests/test_ha.py | 120 +++++++++++++--- tests/test_patroni.py | 7 +- tests/test_postgresql.py | 144 ++++++++++++++++++- tests/test_utils.py | 8 +- 16 files changed, 741 insertions(+), 145 deletions(-) diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 55c7492c..dd42ad28 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -14,6 +14,7 @@ Bootstrap configuration - **loop\_wait**: the number of seconds the loop will sleep. Default value: 10 - **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process. Default value: 30 - **maximum\_lag\_on\_failover**: the maximum bytes a follower may lag to be able to participate in leader election. + - **master\_start\_timeout**: the amount of time a master is allowed to recover from failures before failover is triggered. Default is 300 seconds. When set to 0 failover is done immediately after a crash is detected if possible. When using asynchronous replication a failover can cause lost transactions. Best worst case failover time for master failure is: ttl + master\_start\_timeout + ttl, unless master\_start\_timeout is zero, in which case it's just ttl. Set the value according to your durability/availability tradeoff. - **synchronous\_mode**: turns on synchronous replication mode. In this mode a replica will be chosen as synchronous and only the latest leader and synchronous replica are able to participate in leader election. Synchronous mode makes sure that succesfully committed transactions will not be lost at failover, at the cost of losing availability for writes when Patroni cannot ensure transaction durability. See `replication modes documentation `__ for details. - **postgresql**: - **use\_pg\_rewind**:whether or not to use pg_rewind diff --git a/patroni/__init__.py b/patroni/__init__.py index 8fea4455..d67ac727 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -143,8 +143,30 @@ def patroni_main(): patroni.dcs.delete_leader() +def pg_ctl_start(args): + import subprocess + postmaster = subprocess.Popen(args) + print(postmaster.pid) + + +def call_self(args, **kwargs): + """This function executes Patroni once again with provided arguments. + + :args: list of arguments to call Patroni with. + :returns: `Popen` object""" + + exe = [sys.executable] + if not getattr(sys, 'frozen', False): # Binary distribution? + exe.append(sys.argv[0]) + + import subprocess + return subprocess.Popen(exe + args, **kwargs) + + def main(): if os.getpid() != 1: + if len(sys.argv) > 5 and sys.argv[1] == 'pg_ctl_start': + return pg_ctl_start(sys.argv[2:]) return patroni_main() pid = 0 @@ -173,7 +195,6 @@ def main(): signal.signal(signal.SIGQUIT, passtochild) signal.signal(signal.SIGTERM, passtochild) - import subprocess - patroni = subprocess.Popen([sys.executable] + sys.argv) + patroni = call_self(sys.argv[1:]) pid = patroni.pid patroni.wait() diff --git a/patroni/api.py b/patroni/api.py index 0f1de6e1..56cbd0a4 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -8,7 +8,7 @@ import dateutil.parser import datetime from patroni.exceptions import PostgresConnectionException -from patroni.utils import deep_compare, patch_config, Retry, RetryFailedError, is_valid_pg_version, tzutc +from patroni.utils import deep_compare, patch_config, Retry, RetryFailedError, is_valid_pg_version, parse_int, tzutc from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer from six.moves.socketserver import ThreadingMixIn from threading import Thread @@ -223,6 +223,12 @@ class RestApiHandler(BaseHTTPRequestHandler): status_code = 400 data = "PostgreSQL version should be in the first.major.minor format" break + elif k == 'timeout': + request[k] = parse_int(request[k], 's') + if request[k] is None or request[k] <= 0: + status_code = 400 + data = "Timeout should be a positive number of seconds" + break elif k != 'restart_pending': status_code = 400 data = "Unknown filter for the scheduled restart: {0}".format(k) @@ -236,7 +242,6 @@ class RestApiHandler(BaseHTTPRequestHandler): logger.exception('Exception during restart') status_code = 400 else: - request['postmaster_start_time'] = self.server.patroni.ha.state_handler.postmaster_start_time() if self.server.patroni.ha.schedule_future_restart(request): data = "Restart scheduled" status_code = 202 @@ -379,7 +384,8 @@ class RestApiHandler(BaseHTTPRequestHandler): 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(COALESCE(pg_last_xlog_receive_location(), + pg_last_xlog_replay_location()), '0/0')::bigint, pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint, to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'), pg_is_in_recovery() AND pg_is_xlog_replay_paused(), diff --git a/patroni/config.py b/patroni/config.py index db7a75f7..654b9d10 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -41,6 +41,7 @@ class Config(object): __DEFAULT_CONFIG = { 'ttl': 30, 'loop_wait': 10, 'retry_timeout': 10, 'maximum_lag_on_failover': 1048576, + 'master_start_timeout': 300, 'synchronous_mode': False, 'postgresql': { 'bin_dir': '', diff --git a/patroni/ctl.py b/patroni/ctl.py index 0cedb7db..94b95a33 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -439,9 +439,11 @@ def parse_scheduled(scheduled): @click.option('--pg-version', 'version', help='Restart if the PostgreSQL version is less than provided (e.g. 9.5.2)', default=None) @click.option('--pending', help='Restart if pending', is_flag=True) +@click.option('--timeout', + help='Return error and fail over if necessary when restarting takes longer than this.') @option_force @click.pass_obj -def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, version, pending): +def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, version, pending, timeout): cluster = get_dcs(obj, cluster_name).get_cluster() members = get_members(cluster, cluster_name, member_names, role, force, 'restart') @@ -473,6 +475,9 @@ def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, vers raise PatroniCtlException("Can't schedule restart in the paused state") content['schedule'] = scheduled_at.isoformat() + if timeout is not None: + content['timeout'] = timeout + for member in members: if 'schedule' in content: if force and member.data.get('scheduled_restart'): diff --git a/patroni/ha.py b/patroni/ha.py index 78b94dcc..f17b5331 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -7,6 +7,7 @@ import requests import sys import time +from collections import namedtuple from multiprocessing.pool import ThreadPool from patroni.async_executor import AsyncExecutor from patroni.exceptions import DCSError, PostgresConnectionException @@ -17,6 +18,35 @@ from threading import RLock logger = logging.getLogger(__name__) +class _MemberStatus(namedtuple('_MemberStatus', 'member,reachable,in_recovery,xlog_location,tags')): + """Node status distilled from API response: + + member - dcs.Member object of the node + 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. + is_lagging - `True` if node considers itself too far behind to promote + tags - dictionary with values of different tags (i.e. nofailover) + """ + @classmethod + def from_api_response(cls, member, json): + is_master = json['role'] == 'master' + xlog_location = None if is_master else json['xlog']['received_location'] + return cls(member, True, not is_master, xlog_location, json.get('tags', {})) + + @classmethod + def unknown(cls, member): + return cls(member, False, None, 0, {}) + + def failover_limitation(self): + """Returns reason why this node can't promote or None if everything is ok.""" + if not self.reachable: + return 'not reachable' + if self.tags.get('nofailover', False): + return 'not allowed to promote' + return None + + class Ha(object): def __init__(self, patroni): @@ -26,6 +56,7 @@ class Ha(object): self.cluster = None self.old_cluster = None self.recovering = False + self._start_timeout = None self._async_executor = AsyncExecutor(self.wakeup) # Each member publishes various pieces of information to the DCS using touch_member. This lock protects @@ -140,8 +171,21 @@ class Ha(object): return 'waiting for leader to bootstrap' def recover(self): + if self.has_lock() and self.update_lock(): + timeout = self.patroni.config['master_start_timeout'] + if timeout == 0: + # We are requested to prefer failing over to restarting master. But see first if there + # is anyone to fail over to. + if self.is_failover_possible(self.cluster.members): + logger.info("Master crashed. Failing over.") + self.demote('immediate') + return 'stopped PostgreSQL to fail over after a crash' + else: + timeout = None + self.recovering = True - return self.follow("starting as readonly because i had the session lock", "starting as a secondary", True, True) + return self.follow("starting as readonly because i had the session lock", + "starting as a secondary", True, True, None, timeout) def _get_node_to_follow(self, cluster): # determine the node to follow. If replicatefrom tag is set, @@ -153,7 +197,7 @@ class Ha(object): return node_to_follow if node_to_follow and node_to_follow.name != self.state_handler.name else None - def follow(self, demote_reason, follow_reason, refresh=True, recovery=False, need_rewind=None): + def follow(self, demote_reason, follow_reason, refresh=True, recovery=False, need_rewind=None, timeout=None): if refresh: self.load_cluster_from_dcs() @@ -172,7 +216,8 @@ class Ha(object): elif not node_to_follow: return 'no action' - self.state_handler.follow(node_to_follow, self.cluster.leader, recovery, self._async_executor, need_rewind) + self.state_handler.follow(node_to_follow, self.cluster.leader, recovery, + self._async_executor, need_rewind, timeout) return ret @@ -282,24 +327,16 @@ class Ha(object): @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. - tags - dictionary with values of different tags (i.e. nofailover) + :returns: `_MemberStatus` object """ 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 = None if is_master else json['xlog']['replayed_location'] - return (member, True, not is_master, xlog_location, json.get('tags', {})) + return _MemberStatus.from_api_response(member, response.json()) except Exception as e: logger.warning("request failed: GET %s (%s)", member.api_url, e) - return (member, False, None, 0, {}) + return _MemberStatus.unknown(member) def fetch_nodes_statuses(self, members): pool = ThreadPool(len(members)) @@ -308,23 +345,32 @@ class Ha(object): pool.join() return results + def is_lagging(self, xlog_location): + """Returns if instance with an xlog should consider itself unhealthy to be promoted due to replication lag. + + :param xlog_location: Current xlog location. + :returns True when node is lagging + """ + lag = (self.cluster.last_leader_operation or 0) - xlog_location + return lag > self.state_handler.config.get('maximum_lag_on_failover', 0) + 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 check_replication_lag and not self.state_handler.check_replication_lag(self.cluster.last_leader_operation): + my_xlog_location = self.state_handler.xlog_position() + if check_replication_lag and self.is_lagging(my_xlog_location): 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 not m.nofailover and m.api_url] if members: - my_xlog_location = self.state_handler.xlog_position() - for member, reachable, in_recovery, xlog_location, tags in self.fetch_nodes_statuses(members): - if reachable and not tags.get('nofailover', False): # If the node is unreachable it's not healhy - if not in_recovery: - logger.warning('Master (%s) is still alive', member.name) + for st in self.fetch_nodes_statuses(members): + if st.failover_limitation() is None: + if not st.in_recovery: + logger.warning('Master (%s) is still alive', st.member.name) return False - if my_xlog_location < xlog_location: + if my_xlog_location < st.xlog_location: return False return True @@ -332,13 +378,14 @@ class Ha(object): ret = False members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url] if members: - for member, reachable, _, _, tags in self.fetch_nodes_statuses(members): - if reachable and not tags.get('nofailover', False): - ret = True # TODO: check xlog_location - elif not reachable: - logger.info('Member %s is not reachable', member.name) - elif tags.get('nofailover', False): - logger.info('Member %s is not allowed to promote', member.name) + for st in self.fetch_nodes_statuses(members): + not_allowed_reason = st.failover_limitation() + if not_allowed_reason: + logger.info('Member %s is %s', st.member.name, not_allowed_reason) + elif self.is_lagging(st.xlog_location): + logger.info('Member %s exceeds maximum replication lag', st.member.name) + else: + ret = True else: logger.warning('manual failover: members list is empty') return ret @@ -361,15 +408,13 @@ class Ha(object): # find specific node and check that it is healthy member = self.cluster.get_member(failover.candidate, fallback_to_leader=False) if member: - member, reachable, _, _, tags = self.fetch_node_status(member) - if reachable and not tags.get('nofailover', False): # node is healthy - logger.info('manual failover: to %s, i am %s', member.name, self.state_handler.name) + st = self.fetch_node_status(member) + not_allowed_reason = st.failover_limitation() + if not_allowed_reason is None: # node is healthy + logger.info('manual failover: to %s, i am %s', st.member.name, self.state_handler.name) return False # we wanted to failover to specific member but it is not healthy - if not reachable: - logger.warning('manual failover: member %s is unhealthy', member.name) - elif tags.get('nofailover', False): - logger.warning('manual failover: member %s is not allowed to promote', member.name) + logger.warning('manual failover: member %s is %s', st.member.name, not_allowed_reason) # at this point we should consider all members as a candidates for failover # i.e. we assume that failover.candidate is None @@ -399,6 +444,9 @@ class Ha(object): if ret is not None: # continue if we just deleted the stale failover key as a master return ret + if self.state_handler.is_starting(): # postgresql still starting up is unhealthy + return False + if self.state_handler.is_leader(): # leader is always the healthiest return True @@ -424,18 +472,45 @@ class Ha(object): return self._is_healthiest_node(members.values()) - def demote(self, delete_leader=True): - if delete_leader: - self.state_handler.stop() + def release_leader_key_voluntarily(self): + self.dcs.delete_leader() + self.touch_member() + self.dcs.reset_cluster() + logger.info("Leader key released") + + def demote(self, mode): + """Demote PostgreSQL running as master. + + :param mode: One of offline, graceful or immediate. + offline is used when connection to DCS is not available. + graceful is used when failing over to another node due to user request. May only be called running async. + immediate is used when we determine that we are not suitable for master and want to failover quickly + without regard for data durability. May only be called synchronously. + """ + assert mode in ['offline', 'graceful', 'immediate'] + if mode != 'offline': + if mode == 'immediate': + self.state_handler.stop('immediate', checkpoint=False) + else: + self.state_handler.stop() self.state_handler.set_role('demoted') - self.dcs.delete_leader() - self.dcs.reset_cluster() + self.release_leader_key_voluntarily() time.sleep(2) # Give a time to somebody to take the leader lock cluster = self.dcs.get_cluster() node_to_follow = self._get_node_to_follow(cluster) - return self.state_handler.follow(node_to_follow, cluster.leader, recovery=True, need_rewind=True) + if mode == 'immediate': + # We will try to start up as a standby now. If no one takes the leader lock before we finish + # recovery we will try to promote ourselves. + self._async_executor.schedule('waiting for failover to complete') + self._async_executor.run_async(self.state_handler.follow, + (node_to_follow, cluster.leader, True, None, True)) + else: + return self.state_handler.follow(node_to_follow, cluster.leader, recovery=True, need_rewind=True) else: - self.state_handler.follow(None, None) + # Need to become unavailable as soon as possible, so initiate a stop here. However as we can't release + # the leader key we don't care about confirming the shutdown quickly and can use a regular stop. + self.state_handler.stop(checkpoint=False) + self.state_handler.follow(None, None, recovery=True) def should_run_scheduled_action(self, action_name, scheduled_at, cleanup_fn): if scheduled_at and not self.is_paused(): @@ -454,6 +529,9 @@ class Ha(object): action_name, scheduled_at.isoformat(), delta) return False elif delta < - int(self.dcs.loop_wait * 1.5): + # This means that if run_cycle gets delayed for 2.5x loop_wait we skip the + # scheduled action. Probably not a problem, if things are that bad we don't + # want to be restarting or failing over anyway. logger.warning('Found a stale %s value, cleaning up: %s', action_name, scheduled_at.isoformat()) cleanup_fn() @@ -492,7 +570,7 @@ class Ha(object): if not failover.candidate or m.name == failover.candidate] if self.is_failover_possible(members): # check that there are healthy members self._async_executor.schedule('manual failover: demote') - self._async_executor.run_async(self.demote) + self._async_executor.run_async(self.demote, ('graceful',)) return 'manual failover: demoting myself' else: logger.warning('manual failover: no healthy members found, failover is not possible') @@ -559,7 +637,7 @@ class Ha(object): else: # Either there is no connection to DCS or someone else acquired the lock logger.error('failed to update leader lock') - self.demote(delete_leader=False) + self.demote('offline') return 'demoted self because failed to update leader lock in DCS' else: logger.info('does not have lock') @@ -614,6 +692,7 @@ class Ha(object): def schedule_future_restart(self, restart_data): with self._async_executor: + restart_data['postmaster_start_time'] = self.state_handler.postmaster_start_time() if not self.patroni.scheduled_restart: self.patroni.scheduled_restart = restart_data self.touch_member() @@ -636,10 +715,11 @@ class Ha(object): def restart_scheduled(self): return self._async_executor.scheduled_action == 'restart' - def restart(self, restart_data=None, run_async=False): + def restart(self, restart_data, run_async=False): """ conditional and unconditional restart """ - if (restart_data and isinstance(restart_data, dict) and - not self.restart_matches(restart_data.get('role'), + assert isinstance(restart_data, dict) + + if (not self.restart_matches(restart_data.get('role'), restart_data.get('postgres_version'), ('restart_pending' in restart_data))): return (False, "restart conditions are not satisfied") @@ -649,17 +729,32 @@ class Ha(object): if prev is not None: return (False, prev + ' already in progress') - do_restart = self.state_handler.restart + # Make the main loop to think that we were recovering dead postgres. If we fail + # to start postgres after a specified timeout (see below), we need to remove + # leader key (if it belong to us) rather than trying to start postgres once again. + self.recovering = True + + # No that restart is scheduled we can set timeout for startup, it will get reset + # once async executor runs and main loop notices PostgreSQL as up. + timeout = restart_data.get('timeout', self.patroni.config['master_start_timeout']) + self.set_start_timeout(timeout) + + # For non async cases we want to wait for restart to complete or timeout before returning. + do_restart = functools.partial(self.state_handler.restart, timeout) if self.is_synchronous_mode() and not self.has_lock(): do_restart = functools.partial(self.while_not_sync_standby, do_restart) if run_async: self._async_executor.run_async(do_restart) return (True, 'restart initiated') - elif self._async_executor.run(do_restart): - return (True, 'restarted successfully') else: - return (False, 'restart failed') + res = self._async_executor.run(do_restart) + if res: + return (True, 'restarted successfully') + elif res is None: + return (False, 'postgres is still starting') + else: + return (False, 'restart failed') def _do_reinitialize(self, cluster): self.state_handler.stop('immediate') @@ -712,6 +807,49 @@ class Ha(object): return 'failed to start postgres' return None + def handle_starting_instance(self): + """Starting up PostgreSQL may take a long time. In case we are the leader we may want to + fail over to.""" + + # Check if we are in startup, when paused defer to main loop for manual failovers. + if not self.state_handler.check_for_startup() or self.is_paused(): + self.set_start_timeout(None) + return None + + # state_handler.state == 'starting' here + if self.has_lock(): + if not self.update_lock(): + logger.info("Lost lock while starting up. Demoting self.") + self.demote('immediate') + return 'stopped PostgreSQL while starting up because leader key was lost' + + timeout = self._start_timeout or self.patroni.config['master_start_timeout'] + time_left = timeout - self.state_handler.time_in_state() + + if time_left <= 0: + if self.is_failover_possible(self.cluster.members): + logger.info("Demoting self because master startup is taking too long") + self.demote('immediate') + return 'stopped PostgreSQL because of startup timeout' + else: + return 'master start has timed out, but continuing to wait because failover is not possible' + else: + msg = self.process_manual_failover_from_leader() + if msg is not None: + return msg + + return 'PostgreSQL is still starting up, {0:.0f} seconds until timeout'.format(time_left) + else: + # Use normal processing for standbys + logger.info("Still starting up as a standby.") + return None + + def set_start_timeout(self, value): + """Sets timeout for starting as master before eligible for failover. + + Must be called when async_executor is busy or in the main thread.""" + self._start_timeout = value + def _run_cycle(self): dcs_failed = False try: @@ -731,6 +869,10 @@ class Ha(object): if self._async_executor.busy: return self.handle_long_action_in_progress() + msg = self.handle_starting_instance() + if msg is not None: + return msg + # we've got here, so any async action has finished. if self.recovering and not self.state_handler.need_rewind: self.recovering = False @@ -774,7 +916,7 @@ class Ha(object): # we might not have a valid PostgreSQL connection here if another thread # stops PostgreSQL, therefore, we only reload replication slots if no # asynchronous processes are running (should be always the case for the master) - if not self._async_executor.busy: + if not self._async_executor.busy and not self.state_handler.is_starting(): if not self.state_handler.cb_called: self.state_handler.call_nowait(ACTION_ON_START) self.state_handler.sync_replication_slots(self.cluster) @@ -782,7 +924,7 @@ class Ha(object): dcs_failed = True logger.error('Error communicating with DCS') if not self.is_paused() and self.state_handler.is_running() and self.state_handler.is_leader(): - self.demote(delete_leader=False) + self.demote('offline') return 'demoted self because DCS is not accessible and i was a leader' return 'DCS is not accessible' except (psycopg2.Error, PostgresConnectionException): diff --git a/patroni/postgresql.py b/patroni/postgresql.py index cc596712..b3e60d3e 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -1,4 +1,3 @@ -from collections import defaultdict import logging import os import psycopg2 @@ -9,10 +8,12 @@ import subprocess import tempfile import time +from collections import defaultdict +from patroni import call_self from patroni.exceptions import PostgresConnectionException, PostgresException -from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError +from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop from six import string_types -from threading import Lock +from threading import current_thread, Lock logger = logging.getLogger(__name__) @@ -22,6 +23,11 @@ ACTION_ON_RESTART = "on_restart" ACTION_ON_RELOAD = "on_reload" ACTION_ON_ROLE_CHANGE = "on_role_change" +STATE_RUNNING = 'running' +STATE_REJECT = 'rejecting connections' +STATE_NO_RESPONSE = 'not responding' +STATE_UNKNOWN = 'unknown' + def slot_name_from_member_name(member_name): """Translate member name to valid PostgreSQL slot name. @@ -80,6 +86,7 @@ class Postgresql(object): self._database = config.get('database', 'postgres') self._data_dir = config['data_dir'] self._pending_restart = False + self.__thread_ident = current_thread().ident self._version_file = os.path.join(self._data_dir, 'PG_VERSION') self._major_version = self.get_major_version() @@ -97,6 +104,7 @@ class Postgresql(object): self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass') self.__cb_called = False + self.__cb_pending = None config_base_name = config.get('config_base_name', 'postgresql') self._postgresql_conf = os.path.join(self._data_dir, config_base_name + '.conf') self._postgresql_base_conf_name = config_base_name + '.base.conf' @@ -113,11 +121,17 @@ class Postgresql(object): self.retry = Retry(max_tries=-1, deadline=config['retry_timeout']/2.0, max_delay=1, retry_exceptions=PostgresConnectionException) + # Retry 'pg_is_in_recovery()' only once + self._is_leader_retry = Retry(max_tries=1, deadline=config['retry_timeout']/2.0, max_delay=1, + retry_exceptions=PostgresConnectionException) + self._state_lock = Lock() self.set_state('stopped') self._role_lock = Lock() self.set_role(self.get_postgres_role_from_data_directory()) + self._state_entry_timestamp = None + if self.is_running(): self.set_state('running') self.set_role('master' if self.is_leader() else 'replica') @@ -179,7 +193,7 @@ class Postgresql(object): :returns: `!True` when return_code == 0, otherwise `!False`""" pg_ctl = [self._pgcommand('pg_ctl'), cmd] - if cmd in ('start', 'stop', 'restart'): + if cmd == 'stop': pg_ctl += ['-w'] timeout = self.config.get('pg_ctl_timeout') if timeout: @@ -189,11 +203,31 @@ class Postgresql(object): logger.error('Bad value of pg_ctl_timeout: %s', timeout) return subprocess.call(pg_ctl + ['-D', self._data_dir] + list(args), **kwargs) == 0 + def pg_isready(self): + """Runs pg_isready to see if PostgreSQL is accepting connections. + + :returns: 'ok' if PostgreSQL is up, 'reject' if starting up, 'no_resopnse' if not up.""" + + cmd = [self._pgcommand('pg_isready'), + '-h', self._local_address['host'], + '-p', self._local_address['port'], + '-d', self._database] + # We only need the username because pg_isready does not try to authenticate + if 'username' in self._superuser: + cmd.extend(['-U', self._superuser['username']]) + + ret = subprocess.call(cmd) + return_codes = {0: STATE_RUNNING, + 1: STATE_REJECT, + 2: STATE_NO_RESPONSE, + 3: STATE_UNKNOWN} + return return_codes.get(ret, STATE_UNKNOWN) + def reload_config(self, config): server_parameters = self.get_server_parameters(config) listen_address_changed = pending_reload = pending_restart = False - if self.is_healthy(): + if self.state == 'running': changes = {p: v for p, v in server_parameters.items() if '.' not in p} changes.update({p: None for p, v in self._server_parameters.items() if not ('.' in p or p in changes)}) if changes: @@ -250,7 +284,7 @@ class Postgresql(object): if pending_reload: self._write_postgresql_conf() self.reload() - self.retry.deadline = config['retry_timeout']/2.0 + self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout']/2.0 @property def pending_restart(self): @@ -334,6 +368,9 @@ class Postgresql(object): self._cursor_holder = self._connection = None def _query(self, sql, *params): + """We are always using the same cursor, therefore this method is not thread-safe!!! + You can call it from different threads only if you are holding explicit `AsyncExecutor` lock, + because the main thread is always holding this lock when running HA cycle.""" cursor = None try: cursor = self._cursor() @@ -412,6 +449,9 @@ class Postgresql(object): connstring = 'postgres://{user}@{host}:{port}/{database}'.format(**r) else: connstring = 'postgres://{host}:{port}/{database}'.format(**r) + if 'password' in r: + import getpass + r.setdefault('user', os.environ.get('PGUSER', getpass.getuser())) env = self.write_pgpass(r) if 'password' in r else None try: @@ -521,17 +561,37 @@ class Postgresql(object): return ret def is_leader(self): - return not self.query('SELECT pg_is_in_recovery()').fetchone()[0] + try: + return not self._is_leader_retry(self._query, 'SELECT pg_is_in_recovery()').fetchone()[0] + except RetryFailedError as e: # SELECT pg_is_in_recovery() failed two times + if not self.is_starting() and self.pg_isready() == STATE_REJECT: + self.set_state('starting') + raise PostgresConnectionException(str(e)) def is_running(self): if not (self._version_file_exists() and os.path.isfile(self._postmaster_pid)): return False + return self.is_pid_running(self.read_pid_file().get('pid', 0)) + + def read_pid_file(self): + """Reads and parses postmaster.pid from the data directory + + :returns dictionary of values if successful, empty dictionary otherwise + """ + pid_line_names = ['pid', 'data_dir', 'start_time', 'port', 'socket_dir', 'listen_addr', 'shmem_key'] try: with open(self._postmaster_pid) as f: - pid = int(f.readline()) - if pid < 0: - pid = -pid - return pid > 0 and pid != os.getpid() and pid != os.getppid() and (os.kill(pid, 0) or True) + return {name: line.rstrip("\n") for name, line in zip(pid_line_names, f)} + except IOError: + return {} + + @staticmethod + def is_pid_running(pid): + try: + pid = int(pid) + if pid < 0: + pid = -pid + return pid > 0 and pid != os.getpid() and pid != os.getppid() and (os.kill(pid, 0) or True) except Exception: return False @@ -571,8 +631,48 @@ class Postgresql(object): def set_state(self, value): with self._state_lock: self._state = value + self._state_entry_timestamp = time.time() - def start(self, block_callbacks=False): + def time_in_state(self): + return time.time() - self._state_entry_timestamp + + def is_starting(self): + return self.state == 'starting' + + def wait_for_port_open(self, pid, initiated, timeout): + """Waits until PostgreSQL opens ports.""" + for _ in polling_loop(timeout): + pid_file = self.read_pid_file() + if len(pid_file) > 5: + try: + pmpid = int(pid_file['pid']) + pmstart = int(pid_file['start_time']) + + if pmstart >= initiated - 2 and pmpid == pid: + isready = self.pg_isready() + if isready != STATE_NO_RESPONSE: + if isready not in [STATE_REJECT, STATE_RUNNING]: + logger.warning("Can't determine PostgreSQL startup status, assuming running") + return True + except ValueError: + # Garbage in the pid file + pass + + if not self.is_pid_running(pid): + logger.error('postmaster is not running') + self.set_state('start failed') + return False + + logger.warning("Timed out waiting for PostgreSQL to start") + return False + + def start(self, timeout=None, block_callbacks=False): + """Start PostgreSQL + + Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion + or failure. + + :returns: True if start was initiated and postmaster ports are open, False if start failed""" # make sure we close all connections established against # the former node, otherwise, we might get a stalled one # after kill -9, which would report incorrect data to @@ -583,18 +683,13 @@ class Postgresql(object): logger.error('Cannot start PostgreSQL because one is already running.') return True - self.set_role(self.get_postgres_role_from_data_directory()) - if os.path.exists(self._postmaster_pid): - os.remove(self._postmaster_pid) - logger.info('Removed %s', self._postmaster_pid) - if not block_callbacks: - self.set_state('starting') + self.__cb_pending = ACTION_ON_START - env = {'PATH': os.environ.get('PATH')} - # pg_ctl will write a FATAL if the username is incorrect. exporting PGUSER if necessary - if 'username' in self._superuser and self._superuser['username'] != os.environ.get('USER'): - env['PGUSER'] = self._superuser['username'] + self.set_role(self.get_postgres_role_from_data_directory()) + + self.set_state('starting') + self._pending_restart = False self._write_postgresql_conf() self.resolve_connection_addresses() @@ -602,20 +697,45 @@ class Postgresql(object): opts = {p: self._server_parameters[p] for p, v in self.CMDLINE_OPTIONS.items() if self._major_version >= v[2]} if self._major_version >= 9.6 and opts['wal_level'] == 'hot_standby': opts['wal_level'] = 'replica' - options = ' '.join("--{0}='{1}'".format(p, v) for p, v in opts.items()) + options = ['--{0}={1}'.format(p, v) for p, v in opts.items()] - ret = self.pg_ctl('start', '-o', options, env=env, preexec_fn=os.setsid) - self._pending_restart = False + start_initiated = time.time() - self.set_state('running' if ret else 'start failed') + # Unfortunately `pg_ctl start` does not return postmaster pid to us. Without this information + # it is hard to know the current state of postgres startup, so we had to reimplement pg_ctl start + # in python. It will start postgres, wait for port to be open and wait until postgres will start + # accepting connections. + # Important!!! We can't just start postgres using subprocess.Popen, because in this case it + # will be our child for the rest of our live and we will have to take care of it (`waitpid`). + # So we will use the same approach as pg_ctl uses: start a new process, which will start postgres. + # This process will write postmaster pid to stdout and exit immediately. Now it's responsibility + # of init process to take care about postmaster. + # In order to make everything portable we can't use fork&exec approach here, so we will call + # ourselves and pass list of arguments which must be used to start postgres. + proc = call_self(['pg_ctl_start', self._pgcommand('postgres'), '-D', self._data_dir] + options, close_fds=True, + preexec_fn=os.setsid, stdout=subprocess.PIPE, env={'PATH': os.environ.get('PATH')}) + pid = int(proc.stdout.readline().strip()) + proc.wait() + logger.info('postmaster pid=%s', pid) - self._schedule_load_slots = ret and self.use_slots - self.save_configuration_files() - # block_callbacks is used during restart to avoid - # running start/stop callbacks in addition to restart ones - if ret and not block_callbacks: - self.call_nowait(ACTION_ON_START) - return ret + start_timeout = timeout + if not start_timeout: + try: + start_timeout = float(self.config.get('pg_ctl_timeout', 60)) + except ValueError: + start_timeout = 60 + + # We want postmaster to open ports before we continue + if not self.wait_for_port_open(pid, start_initiated, start_timeout): + return False + + ret = self.wait_for_startup(start_timeout) + if ret is not None: + return ret + elif timeout is not None: + return False + else: + return None def checkpoint(self, connect_kwargs=None): check_not_is_in_recovery = connect_kwargs is not None @@ -642,7 +762,7 @@ class Postgresql(object): self.set_state('stopped') return True - if checkpoint: + if checkpoint and not self.is_starting(): self.checkpoint() if not block_callbacks: @@ -665,12 +785,70 @@ class Postgresql(object): self.call_nowait(ACTION_ON_RELOAD) return ret - def restart(self): - self.set_state('restarting') - ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True) - if ret: - self.call_nowait(ACTION_ON_RESTART) + def check_for_startup(self): + """Checks PostgreSQL status and returns if PostgreSQL is in the middle of startup.""" + return self.is_starting() and not self.check_startup_state_changed() + + def check_startup_state_changed(self): + """Checks if PostgreSQL has completed starting up or failed or still starting. + + Should only be called when state == 'starting' + + :returns: True iff state was changed from 'starting' + """ + ready = self.pg_isready() + + if ready == STATE_REJECT: + return False + elif ready == STATE_NO_RESPONSE: + self.set_state('start failed') + self._schedule_load_slots = False # TODO: can remove this? + self.save_configuration_files() # TODO: maybe remove this? + return True else: + if ready != STATE_RUNNING: + # Bad configuration or unexpected OS error. No idea of PostgreSQL status. + # Let the main loop of run cycle clean up the mess. + logger.warning("%s status returned from pg_isready", + "Unknown" if ready == STATE_UNKNOWN else "Invalid") + self.set_state('running') + self._schedule_load_slots = self.use_slots + self.save_configuration_files() + # TODO: __cb_pending can be None here after PostgreSQL restarts on its own. Do we want to call the callback? + # Previously we didn't even notice. + action = self.__cb_pending or ACTION_ON_START + self.call_nowait(action) + self.__cb_pending = None + + return True + + def wait_for_startup(self, timeout=None): + """Waits for PostgreSQL startup to complete or fail. + + :returns: True if start was successful, False otherwise""" + if not self.is_starting(): + # Should not happen + logger.warning("wait_for_startup() called when not in starting state") + + while not self.check_startup_state_changed(): + if timeout and self.time_in_state() > timeout: + return None + time.sleep(1) + + return self.state == 'running' + + def restart(self, timeout=None): + """Restarts PostgreSQL. + + When timeout parameter is set the call will block either until PostgreSQL has started, failed to start or + timeout arrives. + + :returns: True when restart was successful and timeout did not expire when waiting. + """ + self.set_state('restarting') + self.__cb_pending = ACTION_ON_RESTART + ret = self.stop(block_callbacks=True) and self.start(timeout=timeout, block_callbacks=True) + if not ret and not self.is_starting(): self.set_state('restart failed ({0})'.format(self.state)) return ret @@ -691,9 +869,6 @@ class Postgresql(object): return False return True - def check_replication_lag(self, last_leader_operation): - return (last_leader_operation or 0) - self.xlog_position() <= self.config.get('maximum_lag_on_failover', 0) - def write_pg_hba(self, config): with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'a') as f: f.write('\n{}\n'.format('\n'.join(config))) @@ -804,7 +979,7 @@ class Postgresql(object): def need_rewind(self): return self._need_rewind - def follow(self, member, leader, recovery=False, async_executor=None, need_rewind=None): + def follow(self, member, leader, recovery=False, async_executor=None, need_rewind=None, timeout=None): if need_rewind is not None: self._need_rewind = need_rewind @@ -815,11 +990,11 @@ class Postgresql(object): if async_executor: async_executor.schedule('changing primary_conninfo and restarting') - async_executor.run_async(self._do_follow, (primary_conninfo, leader, recovery)) + async_executor.run_async(self._do_follow, (primary_conninfo, leader, recovery, timeout)) else: - return self._do_follow(primary_conninfo, leader, recovery) + return self._do_follow(primary_conninfo, leader, recovery, timeout) - def _do_follow(self, primary_conninfo, leader, recovery=False): + def _do_follow(self, primary_conninfo, leader, recovery=False, timeout=None): change_role = self.role in ('master', 'demoted') if leader and leader.name == self.name: @@ -830,12 +1005,13 @@ class Postgresql(object): elif change_role: self._need_rewind = True - self._need_rewind &= bool(leader and leader.conn_url) and self.can_rewind + if self._need_rewind and not self.can_rewind: + logger.warning("Data directory may be out of sync master, rewind may be needed.") - if self._need_rewind: + if self._need_rewind and leader and leader.conn_url and self.can_rewind: logger.info("rewind flag is set") - if self.is_running() and not self.stop(): + if self.is_running() and not self.stop(checkpoint=False): return logger.warning('Can not run pg_rewind because postgres is still running') # prepare pg_rewind connection @@ -874,12 +1050,13 @@ class Postgresql(object): else: self.write_recovery_conf(primary_conninfo) if recovery: - self.start() + self.start(timeout=timeout) else: self.restart() self.set_role('replica') if change_role: + # TODO: postpone this until start completes, or maybe do even earlier self.call_nowait(ACTION_ON_ROLE_CHANGE) return True @@ -935,10 +1112,19 @@ $$""".format(name, ' '.join(options)), name, password, password) def xlog_position(self, retry=True): stmt = """SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery() - THEN pg_last_xlog_replay_location() + THEN COALESCE(pg_last_xlog_receive_location(), + pg_last_xlog_replay_location()) ELSE pg_current_xlog_location() END, '0/0')::bigint""" - return (self.query(stmt) if retry else self._query(stmt)).fetchone()[0] + + # This method could be called from different threads (simultaneously with some other `_query` calls). + # If it is called not from main thread we will create a new cursor to execute statement. + if current_thread().ident == self.__thread_ident: + return (self.query(stmt) if retry else self._query(stmt)).fetchone()[0] + + with self.connection().cursor() as cursor: + cursor.execute(stmt) + return cursor.fetchone()[0] def load_replication_slots(self): if self.use_slots and self._schedule_load_slots: @@ -1093,7 +1279,7 @@ $$""".format(name, ' '.join(options)), name, password, password) # TODO: for synchronous_commit = remote_write we actually want to order on write_location for app_name, state, sync_state in self.query( """SELECT application_name, state, sync_state - FROM pg_stat_replication + FROM pg_stat_replication ORDER BY flush_location DESC"""): member = members.get(app_name) if state != 'streaming' or not member or member.tags.get('nosync', False): diff --git a/postgres0.yml b/postgres0.yml index e8c4e8d1..e2794a38 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -22,7 +22,8 @@ bootstrap: loop_wait: 10 retry_timeout: 10 maximum_lag_on_failover: 1048576 - synchronous_mode: false +# master_start_timeout: 300 +# synchronous_mode: false postgresql: use_pg_rewind: true # use_slots: true diff --git a/tests/test_api.py b/tests/test_api.py index 3a76d506..b34b61db 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -63,6 +63,10 @@ class MockHa(object): def schedule_future_restart(data): return True + @staticmethod + def is_lagging(xlog): + return False + @staticmethod def get_effective_tags(): return {'nosync': True} @@ -233,6 +237,10 @@ class TestRestApiHandler(unittest.TestCase): mock_dcs.get_cluster.return_value.is_paused.return_value = True MockRestApiServer(RestApiHandler, make_request(schedule='2016-08-42 12:45TZ+1', role='master')) + # Valid timeout + MockRestApiServer(RestApiHandler, make_request(timeout='60s')) + # Invalid timeout + MockRestApiServer(RestApiHandler, make_request(timeout='42towels')) def test_do_DELETE_restart(self): for retval in (True, False): diff --git a/tests/test_config.py b/tests/test_config.py index f06d1eb0..21d19dae 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -20,9 +20,10 @@ class TestConfig(unittest.TestCase): def test_no_config(self): self.assertRaises(SystemExit, Config) - @patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)) def test_set_dynamic_configuration(self): - self.assertIsNone(self.config.set_dynamic_configuration({'foo': 'bar'})) + with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)): + self.assertIsNone(self.config.set_dynamic_configuration({'foo': 'bar'})) + self.assertTrue(self.config.set_dynamic_configuration({'synchronous_mode': True})) def test_reload_local_configuration(self): os.environ.update({ diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 9c7e457f..645e7a4f 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -237,6 +237,9 @@ class TestCtl(unittest.TestCase): assert 'Error: PostgreSQL version' in result.output assert result.exit_code == 1 + result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force', '--timeout', '10min']) + assert result.exit_code == 0 + with patch('requests.delete', Mock(return_value=MockResponse(500))): # normal restart, the schedule is actually parsed, but not validated in patronictl result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 38293529..0be84cb4 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -42,7 +42,7 @@ def requests_get(url, **kwargs): if url.startswith('http://local'): raise requests.exceptions.RequestException() elif ':8011/patroni' in url: - response.content = '{"role": "replica", "xlog": {"replayed_location": 0}, "tags": {}}' + response.content = '{"role": "replica", "xlog": {"received_location": 0}, "tags": {}}' elif url.endswith('/members'): response.content = '[{}]' if url.startswith('http://error') else members elif url.startswith('http://exhibitor'): diff --git a/tests/test_ha.py b/tests/test_ha.py index eb9265ef..8f8bfc42 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -8,7 +8,7 @@ from patroni.config import Config from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState from patroni.dcs.etcd import Client from patroni.exceptions import DCSError, PostgresException -from patroni.ha import Ha +from patroni.ha import Ha, _MemberStatus from patroni.postgresql import Postgresql from patroni.utils import tzutc from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get @@ -53,6 +53,15 @@ def get_cluster_initialized_with_only_leader(failover=None): l = get_cluster_initialized_without_leader(leader=True, failover=failover).leader return get_cluster(True, l, [l], failover, None) + +def get_node_status(reachable=True, in_recovery=True, xlog_location=10, nofailover=False): + def fetch_node_status(e): + tags = {} + if nofailover: + tags['nofailover'] = True + return _MemberStatus(e, reachable, in_recovery, xlog_location, tags) + return fetch_node_status + future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5) postmaster_start_time = datetime.datetime.now(tzutc) @@ -100,7 +109,7 @@ def run_async(self, func, args=()): @patch.object(Postgresql, 'is_running', Mock(return_value=True)) @patch.object(Postgresql, 'is_leader', Mock(return_value=True)) -@patch.object(Postgresql, 'xlog_position', Mock(return_value=0)) +@patch.object(Postgresql, 'xlog_position', Mock(return_value=10)) @patch.object(Postgresql, 'call_nowait', Mock(return_value=True)) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) @patch.object(Postgresql, 'controldata', Mock(return_value={'Database system identifier': '1234567890'})) @@ -126,6 +135,7 @@ class TestHa(unittest.TestCase): mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) self.p = Postgresql({'name': 'postgresql0', 'scope': 'dummy', 'listen': '127.0.0.1:5432', 'data_dir': 'data/postgresql0', 'retry_timeout': 10, + 'maximum_lag_on_failover': 5, 'authentication': {'superuser': {'username': 'foo', 'password': 'bar'}, 'replication': {'username': '', 'password': ''}}, 'parameters': {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'foo': 'bar', @@ -133,7 +143,6 @@ class TestHa(unittest.TestCase): self.p.set_state('running') self.p.set_role('replica') self.p.postmaster_start_time = MagicMock(return_value=str(postmaster_start_time)) - self.p.check_replication_lag = true self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False) self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test', 'name': 'foo', 'retry_timeout': 10}}) @@ -292,18 +301,20 @@ class TestHa(unittest.TestCase): self.assertIsNotNone(self.ha.reinitialize()) def test_restart(self): - self.assertEquals(self.ha.restart(), (True, 'restarted successfully')) + self.assertEquals(self.ha.restart({}), (True, 'restarted successfully')) + self.p.restart = Mock(return_value=None) + self.assertEquals(self.ha.restart({}), (False, 'postgres is still starting')) self.p.restart = false - self.assertEquals(self.ha.restart(), (False, 'restart failed')) + self.assertEquals(self.ha.restart({}), (False, 'restart failed')) self.ha.cluster = get_cluster_initialized_with_leader() self.ha.reinitialize() - self.assertEquals(self.ha.restart(), (False, 'reinitialize already in progress')) + self.assertEquals(self.ha.restart({}), (False, 'reinitialize already in progress')) with patch.object(self.ha, "restart_matches", return_value=False): self.assertEquals(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied")) def test_restart_in_progress(self): with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)): - self.ha.restart(run_async=True) + self.ha.restart({}, run_async=True) self.assertTrue(self.ha.restart_scheduled()) self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race') @@ -319,6 +330,7 @@ class TestHa(unittest.TestCase): @patch('requests.get', requests_get) @patch('time.sleep', Mock()) def test_manual_failover_from_leader(self): + self.ha.fetch_node_status = get_node_status() self.ha.has_lock = true self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None)) self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') @@ -329,7 +341,9 @@ class TestHa(unittest.TestCase): f = Failover(0, self.p.name, '', None) self.ha.cluster = get_cluster_initialized_with_leader(f) self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself') - self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'}) + self.ha.fetch_node_status = get_node_status(nofailover=True) + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') + self.ha.fetch_node_status = get_node_status(xlog_location=1) self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') # manual failover from the previous leader to us won't happen if we hold the nofailover flag self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None)) @@ -375,17 +389,17 @@ class TestHa(unittest.TestCase): self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None)) self.p.set_role('replica') 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.ha.fetch_node_status = get_node_status() # 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, self.p.name, '', None)) 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, {}) # inaccessible, in_recovery + self.ha.fetch_node_status = get_node_status(reachable=False) # inaccessible, in_recovery self.p.set_role('replica') self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') # set failover flag to True for all members of the cluster # this should elect the current member, as we are not going to call the API for it. self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None)) - self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'}) # accessible, in_recovery + self.ha.fetch_node_status = get_node_status(nofailover=True) # accessible, in_recovery self.p.set_role('replica') self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') # same as previous, but set the current member to nofailover. In no case it should be elected as a leader @@ -409,22 +423,24 @@ class TestHa(unittest.TestCase): def test_is_healthiest_node(self): self.ha.state_handler.is_leader = false self.ha.patroni.nofailover = False - self.ha.fetch_node_status = lambda e: (e, True, True, 0, {}) + self.ha.fetch_node_status = get_node_status() self.assertTrue(self.ha.is_healthiest_node()) + with patch('patroni.postgresql.Postgresql.is_starting', return_value=True): + self.assertFalse(self.ha.is_healthiest_node()) self.ha.is_paused = true self.assertFalse(self.ha.is_healthiest_node()) 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.ha.fetch_node_status = get_node_status() # 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.ha.fetch_node_status = get_node_status(in_recovery=False) # 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.ha.fetch_node_status = get_node_status(xlog_location=11) # accessible, in_recovery, xlog location ahead self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + with patch('patroni.postgresql.Postgresql.xlog_position', return_value=1): + self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.ha.patroni.nofailover = True self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.ha.patroni.nofailover = False @@ -514,6 +530,70 @@ class TestHa(unittest.TestCase): self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) self.assertEquals(self.ha.run_cycle(), 'PAUSE: DCS is not accessible') + @patch('patroni.ha.Ha.update_lock', return_value=True) + @patch('patroni.ha.Ha.demote') + def test_starting_timeout(self, demote, update_lock): + def check_calls(seq): + for mock, called in seq: + if called: + mock.assert_called_once() + else: + mock.assert_not_called() + mock.reset_mock() + self.ha.has_lock = true + self.ha.cluster = get_cluster_initialized_with_leader() + self.p.check_for_startup = true + self.p.time_in_state = lambda: 30 + self.assertEquals(self.ha.run_cycle(), 'PostgreSQL is still starting up, 270 seconds until timeout') + check_calls([(update_lock, True), (demote, False)]) + + self.p.time_in_state = lambda: 350 + self.ha.fetch_node_status = get_node_status(reachable=False) # inaccessible, in_recovery + self.assertEquals(self.ha.run_cycle(), + 'master start has timed out, but continuing to wait because failover is not possible') + check_calls([(update_lock, True), (demote, False)]) + + self.ha.fetch_node_status = get_node_status() # accessible, in_recovery + self.assertEquals(self.ha.run_cycle(), 'stopped PostgreSQL because of startup timeout') + check_calls([(update_lock, True), (demote, True)]) + + update_lock.return_value = False + self.assertEquals(self.ha.run_cycle(), 'stopped PostgreSQL while starting up because leader key was lost') + check_calls([(update_lock, True), (demote, True)]) + + self.ha.has_lock = false + self.p.is_leader = false + self.assertEquals(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') + check_calls([(update_lock, False), (demote, False)]) + + @patch('time.sleep', Mock()) + def test_manual_failover_while_starting(self): + self.ha.has_lock = true + self.p.check_for_startup = true + f = Failover(0, self.p.name, '', None) + self.ha.cluster = get_cluster_initialized_with_leader(f) + self.ha.fetch_node_status = get_node_status() # accessible, in_recovery + self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself') + + @patch('patroni.ha.Ha.demote') + def test_failover_immediately_on_zero_master_start_timeout(self, demote): + self.p.is_running = false + self.ha.cluster = get_cluster_initialized_with_leader() + self.ha.patroni.config.set_dynamic_configuration({'master_start_timeout': 0}) + self.ha.has_lock = true + self.ha.update_lock = true + self.ha.fetch_node_status = get_node_status() # accessible, in_recovery + self.assertEquals(self.ha.run_cycle(), 'stopped PostgreSQL to fail over after a crash') + demote.assert_called_once() + + @patch('time.sleep', Mock()) + @patch('patroni.postgresql.Postgresql.follow') + def test_demote_immediate(self, follow): + self.ha.has_lock = true + self.e.get_cluster = Mock(return_value=get_cluster_initialized_without_leader()) + self.ha.demote('immediate') + follow.assert_called_once_with(None, None, True, None, True) + @patch('time.sleep', Mock()) def test_process_sync_replication(self): self.ha.has_lock = true @@ -648,7 +728,7 @@ class TestHa(unittest.TestCase): get_cluster_initialized_with_leader(sync=('leader', syncstandby)) for syncstandby in ['other', None]]) - self.ha.restart() + self.ha.restart({}) mock_restart.assert_called_once() mock_sleep.assert_called() @@ -656,7 +736,7 @@ class TestHa(unittest.TestCase): # Restart is still called when DCS connection fails mock_restart.reset_mock() self.ha.dcs.get_cluster = Mock(side_effect=DCSError("foo")) - self.ha.restart() + self.ha.restart({}) mock_restart.assert_called_once() @@ -665,7 +745,7 @@ class TestHa(unittest.TestCase): self.ha.dcs.get_cluster.reset_mock() self.ha.touch_member = Mock(return_value=False) - self.ha.restart() + self.ha.restart({}) mock_restart.assert_called_once() self.ha.dcs.get_cluster.assert_not_called() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 5291cd0c..f30ab20d 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -4,7 +4,7 @@ import sys import time import unittest -from mock import Mock, patch +from mock import Mock, PropertyMock, patch from patroni.api import RestApiServer from patroni.async_executor import AsyncExecutor from patroni.dcs.etcd import Client @@ -72,6 +72,10 @@ class TestPatroni(unittest.TestCase): mock_getpid.return_value = 2 _main() + with patch('sys.frozen', Mock(return_value=True), create=True): + sys.argv = ['/patroni', 'pg_ctl_start', 'postgres', '-D', '/data', '--max_connections=100'] + _main() + mock_getpid.return_value = 1 def mock_signal(signo, handler): @@ -98,6 +102,7 @@ class TestPatroni(unittest.TestCase): @patch('patroni.config.Config.save_cache', Mock()) @patch('patroni.config.Config.reload_local_configuration', Mock(return_value=True)) + @patch.object(Postgresql, 'state', PropertyMock(return_value='running')) def test_run(self): self.p.sighup_handler() self.p.ha.dcs.watch = Mock(side_effect=SleepException) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 31b52e70..7253320d 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -8,9 +8,10 @@ import unittest from mock import Mock, MagicMock, PropertyMock, patch, mock_open from patroni.dcs import Cluster, Leader, Member, SyncState from patroni.exceptions import PostgresException, PostgresConnectionException -from patroni.postgresql import Postgresql +from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE from patroni.utils import RetryFailedError from six.moves import builtins +from threading import Thread class MockCursor(object): @@ -204,19 +205,64 @@ class TestPostgresql(unittest.TestCase): def test_delete_trigger_file(self): self.p.delete_trigger_file() + @patch('subprocess.Popen') + @patch.object(Postgresql, 'wait_for_startup') + @patch.object(Postgresql, 'wait_for_port_open') @patch.object(Postgresql, 'is_running') - def test_start(self, mock_is_running): + def test_start(self, mock_is_running, mock_wait_for_port_open, mock_wait_for_startup, mock_popen): mock_is_running.return_value = True + mock_wait_for_port_open.return_value = True + mock_wait_for_startup.return_value = False + mock_popen.stdout.readline.return_value = '123' self.assertTrue(self.p.start()) mock_is_running.return_value = False open(os.path.join(self.data_dir, 'postmaster.pid'), 'w').close() pg_conf = os.path.join(self.data_dir, 'postgresql.conf') open(pg_conf, 'w').close() - self.assertTrue(self.p.start()) + self.assertFalse(self.p.start()) with open(pg_conf) as f: lines = f.readlines() self.assertTrue("f.oo = 'bar'\n" in lines) + mock_wait_for_startup.return_value = None + self.assertFalse(self.p.start(10)) + self.assertIsNone(self.p.start()) + + mock_wait_for_port_open.return_value = False + self.assertFalse(self.p.start()) + + @patch.object(Postgresql, 'pg_isready') + @patch.object(Postgresql, 'read_pid_file') + @patch.object(Postgresql, 'is_pid_running') + @patch('patroni.postgresql.polling_loop', Mock(return_value=range(1))) + def test_wait_for_port_open(self, mock_is_pid_running, mock_read_pid_file, mock_pg_isready): + mock_is_pid_running.return_value = False + mock_pg_isready.return_value = STATE_NO_RESPONSE + + # No pid file and postmaster death + mock_read_pid_file.return_value = {} + self.assertFalse(self.p.wait_for_port_open(42, 100., 1)) + + mock_is_pid_running.return_value = True + + # timeout + mock_read_pid_file.return_value = {'pid', 1} + self.assertFalse(self.p.wait_for_port_open(42, 100., 1)) + + # Garbage pid + mock_read_pid_file.return_value = {'pid': 'garbage', 'start_time': '101', 'data_dir': '', + 'socket_dir': '', 'port': '', 'listen_addr': ''} + self.assertFalse(self.p.wait_for_port_open(42, 100., 1)) + + # Not ready + mock_read_pid_file.return_value = {'pid': '42', 'start_time': '101', 'data_dir': '', + 'socket_dir': '', 'port': '', 'listen_addr': ''} + self.assertFalse(self.p.wait_for_port_open(42, 100., 1)) + + # pg_isready failure + mock_pg_isready.return_value = 'garbage' + self.assertTrue(self.p.wait_for_port_open(42, 100., 1)) + @patch.object(Postgresql, 'is_running') def test_stop(self, mock_is_running): mock_is_running.return_value = True @@ -342,8 +388,11 @@ class TestPostgresql(unittest.TestCase): self.assertRaises(PostgresConnectionException, self.p.query, 'RetryFailedError') self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla') + @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT)) def test_is_leader(self): self.assertTrue(self.p.is_leader()) + with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))): + self.assertRaises(PostgresConnectionException, self.p.is_leader) def test_reload(self): self.assertTrue(self.p.reload()) @@ -362,6 +411,7 @@ class TestPostgresql(unittest.TestCase): def test_last_operation(self): self.assertEquals(self.p.last_operation(), '0') + Thread(target=self.p.last_operation).start() @patch('os.path.isfile', Mock(return_value=True)) @patch('os.kill', Mock(side_effect=Exception)) @@ -385,9 +435,6 @@ 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): @@ -421,12 +468,13 @@ class TestPostgresql(unittest.TestCase): self.assertFalse(self.p.run_bootstrap_post_init({'post_init': '/bin/false'})) with patch('subprocess.call', Mock(return_value=0)) as mock_method: + self.p._superuser.pop('username') self.assertTrue(self.p.run_bootstrap_post_init({'post_init': '/bin/false'})) mock_method.assert_called() args, kwargs = mock_method.call_args assert 'PGPASSFILE' in kwargs['env'].keys() - self.assertEquals(args[0], ['/bin/false', 'postgres://test@localhost:5432/postgres']) + self.assertEquals(args[0], ['/bin/false', 'postgres://localhost:5432/postgres']) @patch('patroni.postgresql.Postgresql.create_replica', Mock(return_value=0)) def test_clone(self): @@ -573,6 +621,88 @@ class TestPostgresql(unittest.TestCase): with patch.object(MockCursor, "execute", side_effect=psycopg2.Error): self.assertIsNone(self.p.postmaster_start_time()) + def test_check_for_startup(self): + with patch('subprocess.call', return_value=0): + self.p._state = 'starting' + self.assertFalse(self.p.check_for_startup()) + self.assertEquals(self.p.state, 'running') + + with patch('subprocess.call', return_value=1): + self.p._state = 'starting' + self.assertTrue(self.p.check_for_startup()) + self.assertEquals(self.p.state, 'starting') + + with patch('subprocess.call', return_value=2): + self.p._state = 'starting' + self.assertFalse(self.p.check_for_startup()) + self.assertEquals(self.p.state, 'start failed') + + with patch('subprocess.call', return_value=0): + self.p._state = 'running' + self.assertFalse(self.p.check_for_startup()) + self.assertEquals(self.p.state, 'running') + + with patch('subprocess.call', return_value=127): + self.p._state = 'running' + self.assertFalse(self.p.check_for_startup()) + self.assertEquals(self.p.state, 'running') + + self.p._state = 'starting' + self.assertFalse(self.p.check_for_startup()) + self.assertEquals(self.p.state, 'running') + + def test_wait_for_startup(self): + state = {'sleeps': 0, 'num_rejects': 0, 'final_return': 0} + + def increment_sleeps(*args): + print("Sleep") + state['sleeps'] += 1 + + def isready_return(*args): + ret = 1 if state['sleeps'] < state['num_rejects'] else state['final_return'] + print("Isready {0} {1}".format(ret, state)) + return ret + + def time_in_state(*args): + return state['sleeps'] + + with patch('subprocess.call', side_effect=isready_return): + with patch('time.sleep', side_effect=increment_sleeps): + self.p.time_in_state = Mock(side_effect=time_in_state) + + self.p._state = 'stopped' + self.assertTrue(self.p.wait_for_startup()) + self.assertEquals(state['sleeps'], 0) + + self.p._state = 'starting' + state['num_rejects'] = 5 + self.assertTrue(self.p.wait_for_startup()) + self.assertEquals(state['sleeps'], 5) + + self.p._state = 'starting' + state['sleeps'] = 0 + state['final_return'] = 2 + self.assertFalse(self.p.wait_for_startup()) + + self.p._state = 'starting' + state['sleeps'] = 0 + state['final_return'] = 0 + self.assertFalse(self.p.wait_for_startup(timeout=2)) + self.assertEquals(state['sleeps'], 3) + + def test_read_pid_file(self): + pidfile = os.path.join(self.data_dir, 'postmaster.pid') + if os.path.exists(pidfile): + os.remove(pidfile) + self.assertEquals(self.p.read_pid_file(), {}) + + @patch('os.kill') + def test_is_pid_running(self, mock_kill): + mock_kill.return_value = True + self.assertTrue(self.p.is_pid_running(-100)) + self.assertFalse(self.p.is_pid_running(0)) + self.assertFalse(self.p.is_pid_running(None)) + def test_pick_sync_standby(self): cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, SyncState(0, self.me.name, self.leadermem.name)) diff --git a/tests/test_utils.py b/tests/test_utils.py index c1144d74..dd84b419 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,7 +2,13 @@ import unittest from mock import Mock, patch from patroni.exceptions import PatroniException -from patroni.utils import Retry, RetryFailedError +from patroni.utils import Retry, RetryFailedError, polling_loop + + +class TestUtils(unittest.TestCase): + + def test_polling_loop(self): + self.assertEquals(list(polling_loop(0.001, interval=0.001)), [0]) @patch('time.sleep', Mock())