Implement scheduled failover.

Scheduled failover allows scheduling of a failover in the future. It does this by writing
a failover key in the DCS which contains the scheduled failover time.
The reason to allow a scheduled failover, is that it does not require one to use a
scheduler (e.g. cron) to schedule such a failover. One of the issues with using
a scheduler is that it may need to authenticate itself. With scheduled failover the authentication
takes place during the scheduling, not during the actual failover.

To allow the time of failover to be expressed, the failover key has changed its format; the old format however
can still be used. The new format expects the failover key to be a json-document with relevant
keys set. We need the timestamp specified to be time zone aware and to be expressed unambigiously, e.g. ISO 8601.

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

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

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:

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]

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<year>\d{4})\-(?P<month>\d{2})\-(?P<day>\d{2})  # date
-T
-(?P<hour>\d{2}):(?P<minute>\d{2}):(?P<second>\d{2})\.(?P<microsecond>\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())