Merge pull request #134 from zalando/feature/scheduled_failover_squashed

Feature/scheduled failover squashed

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
@@ -101,7 +104,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
     @check_auth
     def do_POST_restart(self):
-        status_code = 503
+        status_code = 500
         data = b'restart failed'
         try:
             status, msg = self.server.patroni.ha.restart()
@@ -175,14 +178,34 @@ class RestApiHandler(BaseHTTPRequestHandler):
         leader = request.get('leader')
         member = request.get('member')
         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)
+        status_code = 500
+
+        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'
+                status_code = 422
+        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'
+                    status_code = 503
+                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
@@ -468,10 +470,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.
 
@@ -509,6 +513,25 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
     if candidate and candidate not in candidate_names:
         raise PatroniCtlException('Member {0} does not exist in cluster {1}'.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)
@@ -520,17 +543,14 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
         if not a:
             raise PatroniCtlException('Aborting failover')
 
-    failover_value = '{0}:{1}'.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: {0}'.format(cluster.leader.member.name))
+            logging.debug(cluster)
+            click.echo('{0} {1}'.format(timestamp(), r.text))
         else:
             click.echo('Failover failed, details: {0}, {1}'.format(r.status_code, r.text))
             return
@@ -538,17 +558,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 {0}'.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 {0}'.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)
 
 
@@ -572,10 +584,9 @@ def output_members(cluster, name=None, fmt='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(object):
     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,10 +3,13 @@ import logging
 import psycopg2
 import requests
 import sys
+import datetime
+import pytz
 
+from multiprocessing.pool import ThreadPool
 from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
-from multiprocessing.pool import ThreadPool
+from patroni.utils import sleep
 
 logger = logging.getLogger(__name__)
 
@@ -269,10 +272,34 @@ class Ha(object):
             self.dcs.delete_leader()
             self.touch_member()
             self.dcs.reset_cluster()
-        self.state_handler.follow_the_leader(None)
+        self.state_handler.follow(None)
 
     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)
+            try:
+                delta = (failover.scheduled_at - now).total_seconds()
+
+                if delta > 10:
+                    logging.info('Awaiting failover at %s (in %.0f seconds)', failover.scheduled_at.isoformat(), delta)
+                    return
+                elif delta < -15:
+                    logger.warning('Found a stale failover value, cleaning up: %s', failover.scheduled_at)
+                    self.dcs.manual_failover('', '', self.cluster.failover.index)
+                    return
+
+                # The value is very close to now
+                sleep(max(delta, 0))
+                logger.info('Manual scheduled failover at {}'.format(failover.scheduled_at.isoformat()))
+            except TypeError:
+                logger.warning('Incorrect value in of scheduled_at: %s', failover.scheduled_at)
+
         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())
 
 

requirements-py2.txt

@@ -9,3 +9,5 @@ kazoo>=2.2.1
 python-etcd==0.4.2
 click>=4.1
 prettytable>=0.7
+tzlocal
+python-dateutil

requirements-py3.txt

@@ -9,3 +9,5 @@ kazoo>=2.2.1
 python-etcd==0.4.2
 click>=4.1
 prettytable>=0.7
+tzlocal
+python-dateutil

tests/test_api.py

@@ -176,3 +176,23 @@ class TestRestApiHandler(unittest.TestCase):
         request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
                   b'Content-Length: 50\n\n{"leader": "postgresql1", "member": "postgresql2"}'
         MockRestApiServer(RestApiHandler, request)
+
+        # Valid future date
+        request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
+                  b'"postgresql1", "member": "postgresql2", "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
+        MockRestApiServer(RestApiHandler, request)
+
+        # Exception: No timezone specified
+        request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 97\n\n{"leader": ' +\
+                  b'"postgresql1", "member": "postgresql2", "scheduled_at": "6016-02-15T18:13:30.568224"}'
+        MockRestApiServer(RestApiHandler, request)
+
+        # Exception: Scheduled in the past
+        request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
+                  b'"postgresql1", "member": "postgresql2", "scheduled_at": "1016-02-15T18:13:30.568224+01:00"}'
+        MockRestApiServer(RestApiHandler, request)
+
+        # Invalid date
+        request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
+                  b'"postgresql1", "member": "postgresql2", "scheduled_at": "2010-02-29T18:13:30.568224+01:00"}'
+        MockRestApiServer(RestApiHandler, request)

tests/test_ctl.py

@@ -99,73 +99,95 @@ class TestCtl(unittest.TestCase):
         with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())):
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 other
+
 y''')
-            assert 'Failing over to new leader' in result.output
+            assert 'leader' in result.output
 
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 other
-N''')
-            assert 'Aborting failover' in str(result.output)
+2100-01-01T12:23:00
+y''')
+            assert result.exit_code == 0
 
+            result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+2030-01-01T12:23:00
+y''')
+            assert result.exit_code == 0
+
+            # Aborting failover,as we anser NO to the confirmation
+            result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+
+N''')
+            assert result.exit_code == 1
+
+            # Target and source are equal
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 leader
-y''')
-            assert 'target and source are the same' in str(result.output)
 
+y''')
+            assert result.exit_code == 1
+
+            # Reality is not part of this cluster
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 Reality
+
 y''')
-            assert 'Reality does not exist' in str(result.output)
+            assert result.exit_code == 1
 
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--force'])
-            assert 'Failing over to new leader' in result.output
+            assert 'Member' in result.output
 
+            result = self.runner.invoke(ctl, ['failover', 'dummy', '--force',
+                                              '--scheduled', '2015-01-01T12:00:00+01:00'])
+            assert result.exit_code == 0
+
+            # Invalid timestamp
+            result = self.runner.invoke(ctl, ['failover', 'dummy', '--force', '--scheduled', 'invalid'])
+            assert result.exit_code != 0
+
+            # Invalid timestamp
+            result = self.runner.invoke(ctl, ['failover', 'dummy', '--force',
+                                              '--scheduled', '2115-02-30T12:00:00+01:00'])
+            assert result.exit_code != 0
+
+            # Specifying wrong leader
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='dummy')
-            assert 'is not the leader of cluster' in str(result.output)
+            assert result.exit_code == 1
 
         with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_only_leader())):
+            # No members available
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 other
+
 y''')
-            assert 'No candidates found to failover to' in str(result.output)
+            assert result.exit_code == 1
 
         with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())):
+            # No master available
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 other
+
 y''')
-            assert 'This cluster has no master' in str(result.output)
+            assert result.exit_code == 1
 
         with patch('patroni.ctl.post_patroni', Mock(side_effect=Exception())):
+            # Non-responding patroni
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 other
+
 y''')
             assert 'falling back to DCS' in result.output
-            assert 'Failover failed' in result.output
 
         mocked = Mock()
         mocked.return_value.status_code = 500
         with patch('patroni.ctl.post_patroni', Mock(return_value=mocked)):
             result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
 other
-y''')
-            assert 'Failover failed, details' in result.output
 
-#        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())):
-#            result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='nonsense')
-#            assert 'is not the leader of cluster' in str(result.output)
-#
-#            result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8', '--master', 'nonsense'])
-#            assert 'is not the leader of cluster' in str(result.output)
-#
-#            result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nn')
-#            assert 'Aborting failover' in str(result.output)
-#
-#            with patch('patroni.ctl.wait_for_leader', Mock(return_value = get_cluster_initialized_with_leader())):
-#                result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY')
-#                assert 'master did not change after' in result.output
-#
-#                result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY')
-#                assert 'Failover failed' in result.output
+y''')
+            assert 'Failover failed' in result.output
 
     def test_(self):
         self.assertRaises(patroni.exceptions.PatroniCtlException, get_dcs, {'scheme': 'dummy'}, 'dummy')
@@ -174,6 +196,7 @@ y''')
     @patch('patroni.ctl.query_member', Mock(return_value=([['mock column']], None)))
     def test_query(self):
         with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)):
+            # Mutually exclusive
             result = self.runner.invoke(ctl, [
                 'query',
                 'alpha',
@@ -182,24 +205,13 @@ y''')
                 '--role',
                 'master',
             ])
-            assert 'mutually exclusive' in str(result.output)
+            assert result.exit_code == 1
 
             with self.runner.isolated_filesystem():
                 with open('dummy', 'w') as dummy_file:
                     dummy_file.write('SELECT 1')
 
-                result = self.runner.invoke(ctl, [
-                    'query',
-                    'alpha'
-                ])
-                assert 'You need to specify' in str(result.output)
-
-                result = self.runner.invoke(ctl, [
-                    'query',
-                    'alpha'
-                ])
-                assert 'You need to specify' in str(result.output)
-
+                # Mutually exclusive
                 result = self.runner.invoke(ctl, [
                     'query',
                     'alpha',
@@ -208,7 +220,7 @@ y''')
                     '--command',
                     'dummy',
                 ])
-                assert 'mutually exclusive' in str(result.output)
+                assert result.exit_code == 1
 
                 result = self.runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy'])
 
@@ -217,8 +229,12 @@ y''')
             result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1'])
             assert 'mock column' in result.output
 
-            result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1', '--dbname', 'dummy',
-                                              '--password', '--username', 'dummy'], input='password\n')
+            # --command or --file is mandatory
+            result = self.runner.invoke(ctl, ['query', 'alpha'])
+            assert result.exit_code == 1
+
+            result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1', '--username', 'root',
+                                              '--password', '--dbname', 'postgres'], input='ab\nab')
             assert 'mock column' in result.output
 
     @patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor()))
@@ -248,6 +264,7 @@ y''')
             result = self.runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8'])
             assert 'host=127.0.0.1 port=5435' in result.output
 
+            # Mutually exclusive options
             result = self.runner.invoke(ctl, [
                 'dsn',
                 'alpha',
@@ -256,13 +273,11 @@ y''')
                 '--member',
                 'dummy',
             ])
-            assert 'mutually exclusive' in str(result.output)
+            assert result.exit_code == 1
 
+            # Non-existing member
             result = self.runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy'])
-            assert 'Can not find' in str(result.output)
-
-        # result = self.runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8', '--role', 'replica'])
-        # assert 'host=127.0.0.1 port=5436' in result.output
+            assert result.exit_code == 1
 
     @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
     @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None))
@@ -270,9 +285,16 @@ y''')
     @patch('requests.post', requests_get)
     def test_restart_reinit(self):
         result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
-        result = self.runner.invoke(ctl, ['reinit', 'alpha', '--dcs', '8.8.8.8'], input='y')
+        assert result.exit_code == 0
 
+        result = self.runner.invoke(ctl, ['reinit', 'alpha', '--dcs', '8.8.8.8'], input='y')
+        assert result.exit_code == 1
+
+        # Aborted restart
         result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='N')
+        assert result.exit_code == 1
+
+        # Not a member
         result = self.runner.invoke(ctl, [
             'restart',
             'alpha',
@@ -281,7 +303,7 @@ y''')
             'dummy',
             '--any',
         ], input='y')
-        assert 'not a member' in str(result.output)
+        assert result.exit_code == 1
 
         with patch('requests.post', Mock(return_value=MockResponse())):
             result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
@@ -292,15 +314,18 @@ y''')
         result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nslave')
         assert 'Please confirm' in result.output
         assert 'You are about to remove all' in result.output
-        assert 'You did not exactly type' in str(result.output)
+        # Not typing an exact confirmation
+        assert result.exit_code == 1
 
+        # master specified does not match master of cluster
         result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
 Yes I am aware
 slave''')
-        assert 'You did not specify the current master of the cluster' in str(result.output)
+        assert result.exit_code == 1
 
+        # cluster specified on cmdline does not match verification prompt
         result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader')
-        assert 'Cluster names specified do not match' in str(result.output)
+        assert result.exit_code == 1
 
         with patch('patroni.etcd.Etcd.get_cluster', get_cluster_initialized_with_leader):
             result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'],
@@ -310,11 +335,11 @@ leader''')
             assert 'object has no attribute' in str(result.exception)
 
         with patch('patroni.ctl.get_dcs', Mock(return_value=Mock())):
-            result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'],
-                                        input='''alpha
+            # Not implemented DCS
+            result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
 Yes I am aware
 leader''')
-            assert 'We have not implemented this for DCS of type' in str(result.output)
+            assert result.exit_code == 1
 
     @patch('patroni.etcd.Etcd.watch', Mock(return_value=None))
     @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))

tests/test_etcd.py

@@ -58,10 +58,7 @@ def requests_get(url, **kwargs):
     elif ':8011/patroni' in url:
         response.content = '{"role": "replica", "xlog": {"replayed_location": 0}, "tags": {}}'
     elif url.endswith('/members'):
-        if url.startswith('http://error'):
-            response.content = '[{}]'
-        else:
-            response.content = members
+        response.content = '[{}]' if url.startswith('http://error') else members
     elif url.startswith('http://exhibitor'):
         response.content = '{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}'
     else:

tests/test_ha.py

@@ -1,5 +1,7 @@
 import etcd
 import unittest
+import datetime
+import pytz
 
 from mock import Mock, MagicMock, patch
 from patroni.dcs import Cluster, Failover, Leader, Member
@@ -298,37 +300,58 @@ class TestHa(unittest.TestCase):
     @patch('requests.get', requests_get)
     def test_manual_failover_from_leader(self):
         self.ha.has_lock = true
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', ''))
+        self.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')
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', MockPostgresql.name))
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', MockPostgresql.name, None))
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
-        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla'))
+        self.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')
-        f = Failover(0, MockPostgresql.name, '')
+        f = Failover(0, MockPostgresql.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.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', MockPostgresql.name))
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name, None))
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
 
+        # Failover scheduled time must include timezone
+        scheduled = datetime.datetime.now()
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name, scheduled))
+        self.ha.run_cycle()
+
+        scheduled = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name, scheduled))
+        self.assertEquals('no action.  i am the leader with the lock', self.ha.run_cycle())
+
+        scheduled = scheduled + datetime.timedelta(seconds=30)
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name, scheduled))
+        self.assertEquals('no action.  i am the leader with the lock', self.ha.run_cycle())
+
+        scheduled = scheduled + datetime.timedelta(seconds=-600)
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name, scheduled))
+        self.assertEquals('no action.  i am the leader with the lock', self.ha.run_cycle())
+
+        scheduled = None
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name, scheduled))
+        self.assertEquals('no action.  i am the leader with the lock', self.ha.run_cycle())
+
     @patch('requests.get', requests_get)
     def test_manual_failover_process_no_leader(self):
         self.p.is_leader = false
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', MockPostgresql.name))
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', MockPostgresql.name, None))
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader'))
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None))
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
         self.ha.fetch_node_status = lambda e: (e, True, True, 0, {})  # accessible, in_recovery
         self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
-        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, MockPostgresql.name, ''))
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, MockPostgresql.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.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'))
+        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.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