Manage scheduled restarts using patronictl (#248)

Manage scheduled restarts using patronictl

features/patroni_api.feature

@@ -12,7 +12,7 @@ Scenario: check API requests on a stand-alone server
 	Then I receive a response code 503
 	When I run patronictl.py reinit batman postgres0 --force
 	Then I receive a response returncode 0
-	And I receive a response output "reinitialize failed for member postgres0, status code=503, (I am the leader, can not reinitialize)"
+	And I receive a response output "Failed: reinitialize for member postgres0, status code=503, (I am the leader, can not reinitialize)"
 	When I run patronictl.py failover batman --master postgres0 --force
 	Then I receive a response returncode 1
 	And I receive a response output "Error: No candidates found to failover to"
@@ -54,10 +54,10 @@ Scenario: check API requests for the primary-replica pair
 	And I receive a response role replica
 	When I run patronictl.py reinit batman postgres1 --force
 	Then I receive a response returncode 0
-	And I receive a response output "Succesful reinitialize on member postgres1"
+	And I receive a response output "Success: reinitialize for member postgres1"
 	When I run patronictl.py restart batman postgres0 --force
 	Then I receive a response returncode 0
-	And I receive a response output "Succesful restart on member postgres0"
+	And I receive a response output "Success: restart on member postgres0"
 	And postgres0 role is the primary after 5 seconds
 	When I sleep for 10 seconds
 	Then postgres1 role is the secondary after 15 seconds

patroni/api.py

@@ -7,10 +7,9 @@ import time
 import dateutil.parser
 import datetime
 import pytz
-import re
 
 from patroni.exceptions import PostgresConnectionException
-from patroni.utils import deep_compare, patch_config, Retry, RetryFailedError
+from patroni.utils import deep_compare, patch_config, Retry, RetryFailedError, is_valid_pg_version
 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
 from six.moves.socketserver import ThreadingMixIn
 from threading import Thread
@@ -212,7 +211,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     data = "PostgreSQL role should be either master or replica"
                     break
             elif k == 'postgres_version':
-                if not re.match(r'[1-9][0-9]?(\.(0|([1-9][0-9]?))){2}$', request[k]):
+                if not is_valid_pg_version(request[k]):
                     status_code = 400
                     data = "PostgreSQL version should be in the first.major.minor format"
                     break

patroni/ctl.py

@@ -22,6 +22,7 @@ from patroni.config import Config
 from patroni.dcs import get_dcs as _get_dcs
 from patroni.exceptions import PatroniException
 from patroni.postgresql import get_conn_kwargs
+from patroni.utils import is_valid_pg_version
 from prettytable import PrettyTable
 from six.moves.urllib_parse import urlparse
 
@@ -119,14 +120,16 @@ def auth_header(config):
         return {'Authorization': 'Basic ' + base64.b64encode(config['restapi']['auth'].encode('utf-8')).decode('utf-8')}
 
 
-def post_patroni(member, endpoint, content, headers=None):
+def request_patroni(member, request_type, endpoint, content=None, headers=None):
     headers = headers or {}
-    url = urlparse(member.api_url)
-    logging.debug(url)
+    url_parts = urlparse(member.api_url)
+    logging.debug(url_parts)
     if 'Content-Type' not in headers:
         headers['Content-Type'] = 'application/json'
-    return requests.post('{0}://{1}/{2}'.format(url.scheme, url.netloc, endpoint),
-                         headers=headers,
+
+    url = '{0}://{1}/{2}'.format(url_parts.scheme, url_parts.netloc, endpoint)
+
+    return getattr(requests, request_type)(url, headers=headers,
                                            data=json.dumps(content) if content else None, timeout=60)
 
 
@@ -235,6 +238,37 @@ def get_cursor(cluster, connect_parameters, role='master', member=None):
     return None
 
 
+def get_members(cluster, cluster_name, member_names, role, force, action):
+    candidates = {m.name: m for m in cluster.members}
+
+    if not force or role:
+        output_members(cluster, cluster_name)
+
+    if role:
+        role_names = [m.name for m in get_all_members(cluster, role)]
+        if member_names:
+            member_names = list(set(member_names) & set(role_names))
+            if not member_names:
+                raise PatroniCtlException('No {0} among provided members'.format(role))
+        else:
+            member_names = role_names
+
+    if not member_names and not force:
+        member_names = [click.prompt('Which member do you want to {0} [{1}]?'.format(action,
+                        ', '.join(candidates.keys())), type=str, default='')]
+
+    for mn in member_names:
+        if mn not in candidates:
+            raise PatroniCtlException('{0} is not a member of cluster'.format(mn))
+
+    if not force:
+        confirm = click.confirm('Are you sure you want to {0} members {1}?'.format(action, ', '.join(member_names)))
+        if not confirm:
+            raise PatroniCtlException('Aborted {0}'.format(action))
+
+    return [candidates[n] for n in member_names]
+
+
 @ctl.command('dsn', help='Generate a dsn for the provided member, defaults to a dsn of the master')
 @click.option('--role', '-r', help='Give a dsn of any member with this role', type=click.Choice(['master', 'replica',
               'any']), default=None)
@@ -363,7 +397,7 @@ def query_member(cluster, cursor, member, role, command, connect_parameters):
 def remove(config_file, cluster_name, fmt, dcs):
     _, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
 
-    output_members(cluster, cluster_name, fmt)
+    output_members(cluster, cluster_name, fmt=fmt)
 
     confirm = click.prompt('Please confirm the cluster name to remove', type=str)
     if confirm != cluster_name:
@@ -398,30 +432,6 @@ def wait_for_leader(dcs, timeout=30):
     raise PatroniCtlException('Timeout occured')
 
 
-def empty_post_to_members(cluster, member_names, force, endpoint, headers=None):
-    candidates = {m.name: m for m in cluster.members}
-
-    if not member_names:
-        member_names = [click.prompt('Which member do you want to {0} [{1}]?'.format(endpoint,
-                        ', '.join(candidates.keys())), type=str, default='')]
-
-    for mn in member_names:
-        if mn not in candidates:
-            raise PatroniCtlException('{0} is not a member of cluster'.format(mn))
-
-    if not force:
-        confirm = click.confirm('Are you sure you want to {0} members {1}?'.format(endpoint, ', '.join(member_names)))
-        if not confirm:
-            raise PatroniCtlException('Aborted {0}'.format(endpoint))
-
-    for mn in member_names:
-        r = post_patroni(candidates[mn], endpoint, '', headers)
-        if r.status_code != 200:
-            click.echo('{0} failed for member {1}, status code={2}, ({3})'.format(endpoint, mn, r.status_code, r.text))
-        else:
-            click.echo('Succesful {0} on member {1}'.format(endpoint, mn))
-
-
 def ctl_load_config(cluster_name, config_file, dcs):
     config = load_config(config_file, dcs)
     dcs = get_dcs(config, cluster_name)
@@ -430,31 +440,90 @@ def ctl_load_config(cluster_name, config_file, dcs):
     return config, dcs, cluster
 
 
+def check_response(response, member_name, action_name, silent_success=False):
+    if response.status_code >= 400:
+        click.echo('Failed: {0} for member {1}, status code={2}, ({3})'.format(
+            action_name, member_name, response.status_code, response.text
+        ))
+    elif not silent_success:
+        click.echo('Success: {0} for member {1}'.format(action_name, member_name))
+
+
+def parse_scheduled(scheduled):
+    if (scheduled or 'now') != 'now':
+        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 ({0}). It should be in an unambiguous format (e.g. ISO 8601)'
+            raise PatroniCtlException(message.format(scheduled))
+        return scheduled_at
+
+    return None
+
+
 @ctl.command('restart', help='Restart cluster member')
 @click.argument('cluster_name')
 @click.argument('member_names', nargs=-1)
 @click.option('--role', '-r', help='Restart only members with this role', default='any',
               type=click.Choice(['master', 'replica', 'any']))
 @click.option('--any', 'p_any', help='Restart a single member only', is_flag=True)
+@click.option('--scheduled', help='Timestamp of a scheduled restart in unambiguous format (e.g. ISO 8601)',
+              default=None)
+@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)
 @option_config_file
 @option_force
 @option_dcs
-def restart(cluster_name, member_names, config_file, dcs, force, role, p_any):
+def restart(cluster_name, member_names, config_file, dcs, force, role, p_any, scheduled, version, pending):
     config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
 
-    role_names = [m.name for m in get_all_members(cluster, role)]
-
-    if member_names:
-        member_names = list(set(member_names) & set(role_names))
-    else:
-        member_names = role_names
-
+    members = get_members(cluster, cluster_name, member_names, role, force, 'restart')
     if p_any:
-        random.shuffle(member_names)
-        member_names = member_names[:1]
+        random.shuffle(members)
+        members = members[:1]
 
-    output_members(cluster, cluster_name)
-    empty_post_to_members(cluster, member_names, force, 'restart', auth_header(config))
+    if version is None and not force:
+        version = click.prompt('Restart if the PostgreSQL version is less than provided (e.g. 9.5.2) ',
+                               type=str, default='')
+
+    content = {}
+    if pending:
+        content['restart_pending'] = True
+
+    if version:
+        if not is_valid_pg_version(version):
+            message = 'PostgreSQL version should be in the first.major.minor format'
+            raise PatroniCtlException(message)
+        else:
+            content['postgres_version'] = version
+
+    if scheduled is None and not force:
+        scheduled = click.prompt('When should the restart take place (e.g. 2015-10-01T14:30) ', type=str, default='now')
+
+    scheduled_at = parse_scheduled(scheduled)
+    if scheduled_at:
+        content['schedule'] = scheduled_at.isoformat()
+
+    for member in members:
+        if 'schedule' in content:
+            if force and member.data.get('scheduled_restart'):
+                r = request_patroni(member, 'delete', 'restart', headers=auth_header(config))
+                check_response(r, member.name, 'flush scheduled restart', True)
+
+        r = request_patroni(member, 'post', 'restart', content, auth_header(config))
+        if r.status_code == 200:
+            click.echo('Success: restart on member {0}'.format(member.name))
+        elif r.status_code == 202:
+            click.echo('Success: restart scheduled on member {0}'.format(member.name))
+        elif r.status_code == 409:
+            click.echo('Failed: another restart is already scheduled on member {0}'.format(member.name))
+        else:
+            click.echo('Failed: restart for member {0}, status code={1}, ({2})'.format(
+                member.name, r.status_code, r.text)
+            )
 
 
 @ctl.command('reinit', help='Reinitialize cluster member')
@@ -465,7 +534,11 @@ def restart(cluster_name, member_names, config_file, dcs, force, role, p_any):
 @option_dcs
 def reinit(cluster_name, member_names, config_file, dcs, force):
     config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
-    empty_post_to_members(cluster, member_names, force, 'reinitialize', auth_header(config))
+    members = get_members(cluster, cluster_name, member_names, None, force, 'reinitialize')
+
+    for member in members:
+        r = request_patroni(member, 'post', 'reinitialize', headers=auth_header(config))
+        check_response(r, member.name, 'reinitialize')
 
 
 @ctl.command('failover', help='Failover to a replica')
@@ -474,7 +547,7 @@ def reinit(cluster_name, member_names, config_file, dcs, force):
 @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_force
 @option_config_file
 @option_dcs
 def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled):
@@ -519,16 +592,9 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled
         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 ({0}). It should be in an unambiguous format (e.g. ISO 8601)'
-            raise PatroniCtlException(message.format(scheduled))
+    scheduled_at = parse_scheduled(scheduled)
+
+    if scheduled_at:
         scheduled_at = scheduled_at.isoformat()
 
     failover_value = {'leader': master, 'candidate': candidate, 'scheduled_at': scheduled_at}
@@ -547,7 +613,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled
 
     r = None
     try:
-        r = post_patroni(cluster.leader.member, 'failover', failover_value, auth_header(config))
+        r = request_patroni(cluster.leader.member, 'post', 'failover', failover_value, auth_header(config))
         if r.status_code in (200, 202):
             logging.debug(r)
             cluster = dcs.get_cluster()
@@ -566,7 +632,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled
     output_members(cluster, cluster_name)
 
 
-def output_members(cluster, name, fmt='pretty'):
+def output_members(cluster, name, extended=False, fmt='pretty'):
     rows = []
     logging.debug(cluster)
     leader_name = None
@@ -588,17 +654,28 @@ def output_members(cluster, name, fmt='pretty'):
 
         xlog_location = m.data.get('xlog_location') or 0
         lag = ''
-        if (xlog_location_cluster >= xlog_location):
+        if xlog_location_cluster >= xlog_location:
             lag = round((xlog_location_cluster - xlog_location)/1024/1024)
 
-        rows.append([
+        row = [
             name,
             m.name,
             host,
             leader,
             m.data.get('state', ''),
-            lag
-        ])
+            lag,
+        ]
+        if extended:
+            value = ''
+            scheduled_restart = m.data.get('scheduled_restart')
+            if scheduled_restart:
+                value = scheduled_restart['schedule']
+                if 'postgres_version' in scheduled_restart:
+                    value += ' if version < {0}'.format(scheduled_restart['postgres_version'])
+
+            row.append(value)
+
+        rows.append(row)
 
     columns = [
         'Cluster',
@@ -610,17 +687,22 @@ def output_members(cluster, name, fmt='pretty'):
     ]
     alignment = {'Cluster': 'l', 'Member': 'l', 'Host': 'l', 'Lag in MB': 'r'}
 
+    if extended:
+        columns.append('Scheduled restart')
+        alignment['Scheduled restart'] = 'l'
+
     print_output(columns, rows, alignment, fmt)
 
 
 @ctl.command('list', help='List the Patroni members for a given Patroni')
 @click.argument('cluster_names', nargs=-1)
+@click.option('--extended', '-e', help='Show some extra information', is_flag=True)
 @option_config_file
 @option_format
 @option_watch
 @option_watchrefresh
 @option_dcs
-def members(config_file, cluster_names, fmt, watch, w, dcs):
+def members(config_file, cluster_names, fmt, watch, w, dcs, extended):
     if not cluster_names:
         logging.warning('Listing members: No cluster names were provided')
         return
@@ -630,7 +712,8 @@ def members(config_file, cluster_names, fmt, watch, w, dcs):
         dcs = get_dcs(config, cluster_name)
 
         for _ in watching(w, watch):
-            output_members(dcs.get_cluster(), cluster_name, fmt)
+            cluster = dcs.get_cluster()
+            output_members(cluster, cluster_name, extended, fmt)
 
 
 def timestamp(precision=6):
@@ -646,3 +729,25 @@ def configure(config_file, dcs, namespace):
     config['dcs_api'] = str(dcs)
     config['namespace'] = str(namespace)
     store_config(config, config_file)
+
+
+@ctl.command('flush', help='Flush scheduled events')
+@click.argument('cluster_name')
+@click.argument('member_names', nargs=-1)
+@click.argument('target', type=click.Choice(['restart']))
+@click.option('--role', '-r', help='Flush only members with this role', default='any',
+              type=click.Choice(['master', 'replica', 'any']))
+@option_config_file
+@option_force
+@option_dcs
+def flush(cluster_name, member_names, config_file, dcs, force, role, target):
+    config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
+
+    members = get_members(cluster, cluster_name, member_names, role, force, 'flush')
+    for member in members:
+        if target == 'restart':
+            if member.data.get('scheduled_restart'):
+                r = request_patroni(member, 'delete', 'restart', None, auth_header(config))
+                check_response(r, member.name, 'flush scheduled restart')
+            else:
+                click.echo('No scheduled restart for member {0}'.format(member.name))

patroni/ha.py

@@ -431,6 +431,7 @@ class Ha(object):
         with self._async_executor:
             if not self.patroni.scheduled_restart:
                 self.patroni.scheduled_restart = restart_data
+                self.touch_member()
                 return True
         return False
 
@@ -439,6 +440,7 @@ class Ha(object):
         with self._async_executor:
             if self.patroni.scheduled_restart:
                 self.patroni.scheduled_restart = {}
+                self.touch_member()
                 ret = True
         return ret
 

patroni/utils.py

@@ -2,6 +2,7 @@ import os
 import random
 import sys
 import time
+import re
 
 from patroni.exceptions import PatroniException
 
@@ -226,6 +227,10 @@ def reap_children():
             __reap_children = False
 
 
+def is_valid_pg_version(version):
+    return re.match(r'[1-9][0-9]?(\.(0|([1-9][0-9]?))){2}$', version)
+
+
 class RetryFailedError(PatroniException):
 
     """Raised when retrying an operation ultimately failed, after retrying the maximum number of attempts."""

tests/test_ctl.py

@@ -6,7 +6,7 @@ import unittest
 
 from click.testing import CliRunner
 from mock import patch, Mock
-from patroni.ctl import ctl, members, store_config, load_config, output_members, post_patroni, get_dcs, parse_dcs, \
+from patroni.ctl import ctl, members, store_config, load_config, output_members, request_patroni, get_dcs, parse_dcs, \
     wait_for_leader, get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException
 from psycopg2 import OperationalError
 from test_etcd import etcd_read, requests_get, socket_getaddrinfo, MockResponse
@@ -66,7 +66,7 @@ class TestCtl(unittest.TestCase):
         self.assertIsNone(output_members(cluster, name='abc', fmt='tsv'))
 
     @patch('patroni.ctl.get_dcs')
-    @patch('patroni.ctl.post_patroni', Mock(return_value=MockResponse()))
+    @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse()))
     def test_failover(self, mock_get_dcs):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
@@ -109,12 +109,12 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['failover', 'dummy'], input='dummy')
         assert result.exit_code == 1
 
-        with patch('patroni.ctl.post_patroni', Mock(side_effect=Exception)):
+        with patch('patroni.ctl.request_patroni', Mock(side_effect=Exception)):
             # Non-responding patroni
             result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n\ny')
             assert 'falling back to DCS' in result.output
 
-        with patch('patroni.ctl.post_patroni') as mocked:
+        with patch('patroni.ctl.request_patroni') as mocked:
             mocked.return_value.status_code = 500
             result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n\ny')
             assert 'Failover failed' in result.output
@@ -205,23 +205,72 @@ class TestCtl(unittest.TestCase):
     @patch('patroni.ctl.get_dcs')
     def test_restart_reinit(self, mock_get_dcs):
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
-        result = self.runner.invoke(ctl, ['restart', 'alpha'], input='y')
-        assert 'restart failed for' in result.output
+        result = self.runner.invoke(ctl, ['restart', 'alpha'], input='y\n\nnow')
+        assert 'Failed: restart for' in result.output
         assert result.exit_code == 0
 
         result = self.runner.invoke(ctl, ['reinit', 'alpha'], input='y')
         assert result.exit_code == 1
 
+        # successful reinit
+        result = self.runner.invoke(ctl, ['reinit', 'alpha', 'other'], input='y')
+        assert result.exit_code == 0
+
         # Aborted restart
         result = self.runner.invoke(ctl, ['restart', 'alpha'], input='N')
         assert result.exit_code == 1
 
+        result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force'])
+        assert result.exit_code == 0
+
         # Not a member
         result = self.runner.invoke(ctl, ['restart', 'alpha', 'dummy', '--any'], input='y')
         assert result.exit_code == 1
 
+        # Wrong pg version
+        result = self.runner.invoke(ctl, ['restart', 'alpha', '--any', '--pg-version', '9.1'], input='y')
+        assert 'Error: PostgreSQL version' in result.output
+        assert result.exit_code == 1
+
+        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', '--scheduled', '2300-10-01T14:30'])
+            assert 'Failed: flush scheduled restart' in result.output
+
         with patch('requests.post', Mock(return_value=MockResponse())):
-            result = self.runner.invoke(ctl, ['restart', 'alpha'], input='y')
+            # normal restart, the schedule is actually parsed, but not validated in patronictl
+            result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0.0',
+                                        '--scheduled', '2300-10-01T14:30'], input='y')
+            assert result.exit_code == 0
+
+        with patch('requests.post', Mock(return_value=MockResponse(204))):
+            # get restart with the non-200 return code
+            # normal restart, the schedule is actually parsed, but not validated in patronictl
+            result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0.0',
+                                        '--scheduled', '2300-10-01T14:30'], input='y')
+            assert result.exit_code == 0
+
+        # force restart with restart already present
+        with patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse(204))):
+            result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
+            assert result.exit_code == 0
+
+        with patch('requests.post', Mock(return_value=MockResponse(202))):
+            # get restart with the non-200 return code
+            # normal restart, the schedule is actually parsed, but not validated in patronictl
+            result = self.runner.invoke(
+                ctl, ['restart', 'alpha', '--pg-version', '99.0.0', '--scheduled', '2300-10-01T14:30'], input='y'
+            )
+            assert 'Success: restart scheduled' in result.output
+            assert result.exit_code == 0
+
+        with patch('requests.post', Mock(return_value=MockResponse(409))):
+            # get restart with the non-200 return code
+            # normal restart, the schedule is actually parsed, but not validated in patronictl
+            result = self.runner.invoke(
+                ctl, ['restart', 'alpha', '--pg-version', '99.0.0', '--scheduled', '2300-10-01T14:30'], input='y'
+            )
+            assert 'Failed: another restart is already' in result.output
             assert result.exit_code == 0
 
     @patch('patroni.ctl.get_dcs')
@@ -253,9 +302,9 @@ class TestCtl(unittest.TestCase):
         assert cluster.leader.member.name == 'leader'
 
     @patch('requests.post', Mock(side_effect=requests.exceptions.ConnectionError('foo')))
-    def test_post_patroni(self):
+    def test_request_patroni(self):
         member = get_cluster_initialized_with_leader().leader.member
-        self.assertRaises(requests.exceptions.ConnectionError, post_patroni, member, 'dummy', {})
+        self.assertRaises(requests.exceptions.ConnectionError, request_patroni, member, 'post', 'dummy', {})
 
     def test_ctl(self):
         self.runner.invoke(ctl, ['list'])
@@ -292,3 +341,27 @@ class TestCtl(unittest.TestCase):
     def test_configure(self):
         result = self.runner.invoke(configure, ['--dcs', 'abc', '-c', 'dummy', '-n', 'bla'])
         assert result.exit_code == 0
+
+    @patch('patroni.ctl.get_dcs')
+    def test_list_extended(self, mock_get_dcs):
+        mock_get_dcs.return_value = self.e
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
+
+        result = self.runner.invoke(ctl, ['list', 'dummy', '--extended'])
+        assert '2100' in result.output
+        assert 'Scheduled restart' in result.output
+
+    @patch('patroni.ctl.get_dcs')
+    @patch('requests.delete', Mock(return_value=MockResponse()))
+    def test_flush(self, mock_get_dcs):
+        mock_get_dcs.return_value = self.e
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
+
+        result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '-r', 'master'], input='y')
+        assert 'No scheduled restart' in result.output
+
+        result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force'])
+        assert 'Success: flush scheduled restart' in result.output
+        with patch.object(requests, 'delete', return_value=MockResponse(404)):
+            result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force'])
+            assert 'Failed: flush scheduled restart' in result.output

tests/test_etcd.py

@@ -13,8 +13,8 @@ from urllib3.exceptions import ReadTimeoutError
 
 class MockResponse(object):
 
-    def __init__(self):
-        self.status_code = 200
+    def __init__(self, status_code=200):
+        self.status_code = status_code
         self.content = '{}'
         self.ok = True
         self.text = ''

tests/test_ha.py

@@ -34,7 +34,10 @@ def get_cluster_initialized_without_leader(leader=False, failover=None):
                                   'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4})
     l = Leader(0, 0, m1) if leader else None
     m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
-                                 'api_url': 'http://127.0.0.1:8011/patroni', 'tags': {'clonefrom': True}})
+                                 'api_url': 'http://127.0.0.1:8011/patroni',
+                                 'tags': {'clonefrom': True},
+                                 'scheduled_restart': {'schedule': "2100-01-01 10:53:07.560445+00:00",
+                                                       'postgres_version': '99.0.0'}})
     return get_cluster(True, l, [m1, m2], failover)
 
 
@@ -407,8 +410,8 @@ class TestHa(unittest.TestCase):
     def test_schedule_future_restart(self):
         self.ha.patroni.scheduled_restart = {}
         # do the restart 2 times. The first one should succeed, the second one should fail
-        self.assertTrue(self.ha.schedule_future_restart({'schedule': str(future_restart_time)}))
-        self.assertFalse(self.ha.schedule_future_restart({'schedule': str(future_restart_time)}))
+        self.assertTrue(self.ha.schedule_future_restart({'schedule': future_restart_time}))
+        self.assertFalse(self.ha.schedule_future_restart({'schedule': future_restart_time}))
 
     def test_delete_future_restarts(self):
         self.ha.delete_future_restart()