diff --git a/patroni/api.py b/patroni/api.py index 2318fbf0..677c8f85 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -5,6 +5,9 @@ import logging import psycopg2 import socket import time +import dateutil +import datetime +import pytz from patroni.exceptions import PostgresConnectionException from patroni.utils import Retry, RetryFailedError @@ -176,13 +179,31 @@ class RestApiHandler(BaseHTTPRequestHandler): member = request.get('member', None) cluster = self.server.patroni.ha.dcs.get_cluster() status_code = 503 - data = self.is_failover_possible(cluster, leader, member) - if not data: - if not self.server.patroni.dcs.manual_failover(leader, member): - data = b'failed to write failover key into DCS' - else: - self.server.patroni.dcs.event.set() - status_code, data = self.poll_failover_result(cluster.leader and cluster.leader.name, member) + + data = b'' + if request.get('scheduled_at'): + try: + scheduled_at = dateutil.parser.parse(request['scheduled_at']) + if scheduled_at.tzinfo is None: + data = b'Timezone information is mandatory for scheduled_at' + status_code = 400 + elif scheduled_at < datetime.datetime.now(pytz.utc): + data = b'Cannot schedule failover in the past' + status_code = 422 + elif self.server.patroni.dcs.manual_failover(leader, member, scheduled_at): + data = b'Failover scheduled' + status_code = 200 + except (ValueError, TypeError): + logger.exception('Invalid scheduled failover time: {}'.format(request['scheduled_at'])) + data = b'Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601' + else: + data = self.is_failover_possible(cluster, leader, member) + if not data: + if not self.server.patroni.dcs.manual_failover(leader, member): + data = b'failed to write failover key into DCS' + else: + self.server.patroni.dcs.event.set() + status_code, data = self.poll_failover_result(cluster.leader and cluster.leader.name, member) self.send_response(status_code) self.send_header('Content-Type', 'text/html') diff --git a/patroni/ctl.py b/patroni/ctl.py index 635b0a89..1d923420 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -14,6 +14,8 @@ import datetime from prettytable import PrettyTable from six.moves.urllib_parse import urlparse import logging +import dateutil +import tzlocal from .etcd import Etcd from .exceptions import PatroniCtlException @@ -473,10 +475,12 @@ def reinit(cluster_name, member_names, config_file, dcs, force): @click.argument('cluster_name') @click.option('--master', help='The name of the current master', default=None) @click.option('--candidate', help='The name of the candidate', default=None) +@click.option('--scheduled', help='Timestamp of a scheduled failover in unambiguous format (e.g. ISO 8601)', + default=None) @click.option('--force', is_flag=True) @option_config_file @option_dcs -def failover(config_file, cluster_name, master, candidate, force, dcs): +def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled): """ We want to trigger a failover for the specified cluster name. @@ -514,6 +518,25 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): if candidate and candidate not in candidate_names: raise PatroniCtlException('Member {} does not exist in cluster {}'.format(candidate, cluster_name)) + if scheduled is None and not force: + scheduled = click.prompt('When should the failover take place (e.g. 2015-10-01T14:30) ', type=str, + default='now') + + if (scheduled or 'now') == 'now': + scheduled_at = None + else: + try: + scheduled_at = dateutil.parser.parse(scheduled) + if scheduled_at.tzinfo is None: + scheduled_at = tzlocal.get_localzone().localize(scheduled_at) + except (ValueError, TypeError): + message = 'Unable to parse scheduled timestamp ({}). It should be in an unambiguous format (e.g. ISO 8601)' + raise PatroniCtlException(message.format(scheduled)) + scheduled_at = scheduled_at.isoformat() + + failover_value = {'leader': master, 'member': candidate, 'scheduled_at': scheduled_at} + logging.debug(failover_value) + # By now we have established that the leader exists and the candidate exists click.echo('Current cluster topology') output_members(dcs.get_cluster(), name=cluster_name) @@ -525,17 +548,14 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): if not a: raise PatroniCtlException('Aborting failover') - failover_value = '{}:{}'.format(master, candidate or '') - - t_started = time.time() r = None try: - r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'member': candidate or ''}) + r = post_patroni(cluster.leader.member, 'failover', failover_value) if r.status_code == 200: logging.debug(r) - logging.debug(r.text) cluster = dcs.get_cluster() - click.echo(timestamp() + ' Failing over to new leader: {}'.format(cluster.leader.member.name)) + logging.debug(cluster) + click.echo('{} {}'.format(timestamp(), r.text)) else: click.echo('Failover failed, details: {}, {}'.format(r.status_code, r.text)) return @@ -543,17 +563,9 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): logging.exception(r) logging.warning('Failing over to DCS') click.echo(timestamp() + ' Could not failover using Patroni api, falling back to DCS') - dcs.set_failover_value(failover_value) - click.echo(timestamp() + ' Initialized failover from master {}'.format(master)) - # The failover process should within a minute update the failover key, we will keep watching it until it changes - # or we timeout - cluster = wait_for_leader(dcs, timeout=60) - if cluster.leader.member.name == master: - click.echo('Failover failed, master did not change after {:0.1f} seconds'.format(time.time() - t_started)) - return + click.echo(timestamp() + ' Initializing failover from master {}'.format(master)) + dcs.manual_failover(leader=master, member=candidate, scheduled_at=failover_value) - click.echo(timestamp() + ' Failover completed in {:0.1f} seconds, new leader is {}'.format(time.time() - t_started, - str(cluster.leader.member.name))) output_members(cluster, name=cluster_name) @@ -577,10 +589,9 @@ def output_members(cluster, name=None, format='pretty'): host = build_connect_parameters(m.conn_url)['host'] - xlog_location = m.data.get('xlog_location') - if xlog_location is None or (xlog_location_cluster < xlog_location): - lag = '' - else: + xlog_location = m.data.get('xlog_location') or 0 + lag = '' + if (xlog_location_cluster >= xlog_location): lag = round((xlog_location_cluster - xlog_location)/1024/1024) rows.append([ diff --git a/patroni/dcs.py b/patroni/dcs.py index a6b9d856..c689c4a9 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -1,5 +1,6 @@ import abc import json +import dateutil from collections import namedtuple from patroni.exceptions import DCSError @@ -89,12 +90,44 @@ class Leader(namedtuple('Leader', 'index,session,member')): return self.member.conn_url -class Failover(namedtuple('Failover', 'index,leader,member')): +class Failover(namedtuple('Failover', 'index,leader,member,scheduled_at')): + """ + >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader"}')) + True + >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader", "member": "cluster:member"}')) + True + >>> Failover.from_node(1, 'null') is None + True + >>> n = '{"leader": "cluster_leader", "member": "cluster:member", "scheduled_at": "2016-01-14T10:09:57.1394Z"}' + >>> 'tzinfo=' in str(Failover.from_node(1, n)) + True + >>> Failover.from_node(1, None) is None + True + >>> Failover.from_node(1, '{}') is None + True + >>> 'abc' in Failover.from_node(1, 'abc:def') + True + """ @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 + if not value: + return None + + try: + data = json.loads(value) + if not data: + return None + except ValueError: + t = [a.strip() for a in value.split(':')] + leader = t[0] + candidate = t[1] if len(t) > 1 else None + return Failover(index, leader, candidate, None) if leader or candidate else None + + if data.get('scheduled_at'): + data['scheduled_at'] = dateutil.parser.parse(data['scheduled_at']) + + return Failover(index, data.get('leader'), data.get('member'), data.get('scheduled_at')) class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members,failover')): @@ -223,8 +256,18 @@ class AbstractDCS: 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 manual_failover(self, leader, member, scheduled_at=None, index=None): + failover_value = dict() + if leader: + failover_value['leader'] = leader + + if member: + failover_value['member'] = member + + if scheduled_at: + failover_value['scheduled_at'] = scheduled_at.isoformat() + + return self.set_failover_value(json.dumps(failover_value), index) def current_leader(self): try: diff --git a/patroni/ha.py b/patroni/ha.py index 7e78eb77..dc038da9 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -3,6 +3,9 @@ import logging import psycopg2 import requests import sys +import time +import datetime +import pytz from patroni.async_executor import AsyncExecutor from patroni.exceptions import DCSError, PostgresConnectionException @@ -268,6 +271,28 @@ class Ha: def process_manual_failover_from_leader(self): failover = self.cluster.failover + + if failover.scheduled_at: + # If the failover is in the far future, we shouldn't do anything and just return. + # If the failover is in the past, we consider the value to be stale and we remove + # the value. + # If the value is close to now, we initiate the failover + now = datetime.datetime.now(pytz.utc) + delta = (failover.scheduled_at - now).total_seconds() + + if delta > 10: + logging.info('Awaiting failover at {0} (in {1:.0f} seconds)'.format(failover.scheduled_at.isoformat(), + delta)) + return + elif delta < -15: + logger.warning('Found a stale failover value, cleaning up: {}'.format(failover.scheduled_at)) + self.dcs.manual_failover('', '', self.cluster.failover.index) + return + + # The value is very close to now + time.sleep(max(delta, 0)) + logger.info('Manual scheduled failover at {}'.format(failover.scheduled_at.isoformat())) + 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] diff --git a/patroni/utils.py b/patroni/utils.py index 9b040294..7e827222 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -1,10 +1,11 @@ import datetime import os import random -import re import signal import sys import time +import pytz +import dateutil.parser from patroni.exceptions import PatroniException @@ -12,39 +13,23 @@ ignore_sigterm = False interrupted_sleep = False reap_children = False -_DATE_TIME_RE = re.compile(r'''^ -(?P\d{4})\-(?P\d{2})\-(?P\d{2}) # date -T -(?P\d{2}):(?P\d{2}):(?P\d{2})\.(?P\d{6}) # time -\d*Z$''', re.X) - - -def parse_datetime(time_str): - """ - >>> parse_datetime('2015-06-10T12:56:30.552539016Z') - datetime.datetime(2015, 6, 10, 12, 56, 30, 552539) - >>> parse_datetime('2015-06-10 12:56:30.552539016Z') - """ - m = _DATE_TIME_RE.match(time_str) - if not m: - return None - p = dict((n, int(m.group(n))) for n in 'year month day hour minute second microsecond'.split(' ')) - return datetime.datetime(**p) - def calculate_ttl(expiration): """ >>> calculate_ttl(None) - >>> calculate_ttl('2015-06-10 12:56:30.552539016Z') + >>> calculate_ttl('2015-06-10 12:56:30.552539016Z') < 0 + True >>> calculate_ttl('2015-06-10T12:56:30.552539016Z') < 0 True + >>> calculate_ttl('fail-06-10T12:56:30.552539016Z') """ if not expiration: return None - expiration = parse_datetime(expiration) - if not expiration: + try: + expiration = dateutil.parser.parse(expiration) + except (ValueError, TypeError): return None - now = datetime.datetime.utcnow() + now = datetime.datetime.now(pytz.utc) return int((expiration - now).total_seconds())