From 3a96ffa7189efba2241aa11e940fb781b5cf419a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 5 Jan 2018 15:16:45 +0100 Subject: [PATCH] Expose pause state of every member to DCS and via REST (#592) and implement patronictl pause|resume --wait on top of that Fixes https://github.com/zalando/patroni/issues/349 --- patroni/api.py | 2 ++ patroni/ctl.py | 40 ++++++++++++++++++++++++++++++++-------- patroni/ha.py | 3 +++ tests/test_api.py | 4 ++++ tests/test_ctl.py | 18 +++++++++++++----- tests/test_ha.py | 1 + 6 files changed, 55 insertions(+), 13 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 2bb43ad4..82406833 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -70,6 +70,8 @@ class RestApiHandler(BaseHTTPRequestHandler): response['scheduled_restart']['schedule'] = (response['scheduled_restart']['schedule']).isoformat() if not patroni.ha.watchdog.is_healthy: response['watchdog_failed'] = True + if patroni.ha.is_paused(): + response['pause'] = True self._write_json_response(status_code, response) def do_GET(self, write_status_code_only=False): diff --git a/patroni/ctl.py b/patroni/ctl.py index 4c1f0758..496c9fe6 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -30,7 +30,7 @@ from patroni.config import Config from patroni.dcs import get_dcs as _get_dcs from patroni.exceptions import PatroniException from patroni.postgresql import Postgresql -from patroni.utils import patch_config +from patroni.utils import patch_config, polling_loop from patroni.version import __version__ from prettytable import PrettyTable from six.moves.urllib_parse import urlparse @@ -815,8 +815,27 @@ def flush(obj, cluster_name, member_names, force, role, target): click.echo('No scheduled restart for member {0}'.format(member.name)) -def toggle_pause(config, cluster_name, paused): - cluster = get_dcs(config, cluster_name).get_cluster() +def wait_until_pause_is_applied(dcs, paused, old_cluster): + click.echo("'{0}' request sent, waiting until it is recognized by all nodes".format(paused and 'pause' or 'resume')) + old = {m.name: m.index for m in old_cluster.members if m.api_url} + loop_wait = old_cluster.config.data.get('loop_wait', dcs.loop_wait) + + for _ in polling_loop(loop_wait + 1): + cluster = dcs.get_cluster() + if all(m.data.get('pause', False) == paused for m in cluster.members if m.name in old): + break + else: + remaining = [m.name for m in cluster.members if m.data.get('pause', False) != paused + and m.name in old and old[m.name] != m.index] + if remaining: + return click.echo("{0} members didn't recognized pause state after {1} seconds" + .format(', '.join(remaining), loop_wait)) + return click.echo('Success: cluster management is {0}'.format(paused and 'paused' or 'resumed')) + + +def toggle_pause(config, cluster_name, paused, wait): + dcs = get_dcs(config, cluster_name) + cluster = dcs.get_cluster() if cluster.is_paused() == paused: raise PatroniCtlException('Cluster is {0} paused'.format(paused and 'already' or 'not')) @@ -833,7 +852,10 @@ def toggle_pause(config, cluster_name, paused): continue if r.status_code == 200: - click.echo('Success: cluster management is {0}'.format(paused and 'paused' or 'resumed')) + if wait: + wait_until_pause_is_applied(dcs, paused, cluster) + else: + click.echo('Success: cluster management is {0}'.format(paused and 'paused' or 'resumed')) else: click.echo('Failed: {0} cluster management status code={1}, ({2})'.format( paused and 'pause' or 'resume', r.status_code, r.text)) @@ -845,15 +867,17 @@ def toggle_pause(config, cluster_name, paused): @ctl.command('pause', help='Disable auto failover') @arg_cluster_name @click.pass_obj -def pause(obj, cluster_name): - return toggle_pause(obj, cluster_name, True) +@click.option('--wait', help='Wait until pause is applied on all nodes', is_flag=True) +def pause(obj, cluster_name, wait): + return toggle_pause(obj, cluster_name, True, wait) @ctl.command('resume', help='Resume auto failover') @arg_cluster_name +@click.option('--wait', help='Wait until pause is cleared on all nodes', is_flag=True) @click.pass_obj -def resume(obj, cluster_name): - return toggle_pause(obj, cluster_name, False) +def resume(obj, cluster_name, wait): + return toggle_pause(obj, cluster_name, False, wait) @contextmanager diff --git a/patroni/ha.py b/patroni/ha.py index f5ebaefe..65845a89 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -133,6 +133,9 @@ class Ha(object): scheduled_restart_data['schedule'] = scheduled_restart_data['schedule'].isoformat() data['scheduled_restart'] = scheduled_restart_data + if self.is_paused(): + data['pause'] = True + return self.dcs.touch_member(data) def clone(self, clone_member=None, msg='(without leader)'): diff --git a/tests/test_api.py b/tests/test_api.py index 90d84f9d..1bc40cdf 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -83,6 +83,10 @@ class MockHa(object): def wakeup(): pass + @staticmethod + def is_paused(): + return True + class MockPatroni(object): diff --git a/tests/test_ctl.py b/tests/test_ctl.py index deafdc96..0cda4baf 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -15,7 +15,7 @@ from patroni.utils import tzutc from psycopg2 import OperationalError from test_etcd import etcd_read, requests_get, socket_getaddrinfo, MockResponse from test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, \ - get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader + get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader, get_cluster, Member from test_postgresql import MockConnect, psycopg2_connect CONFIG_FILE_PATH = './test-ctl.yaml' @@ -406,14 +406,11 @@ class TestCtl(unittest.TestCase): assert 'Failed: flush scheduled restart' in result.output @patch('patroni.ctl.get_dcs') + @patch('patroni.ctl.polling_loop', Mock(return_value=[1])) def test_pause_cluster(self, mock_get_dcs): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - with patch('requests.patch', Mock(return_value=MockResponse(200))): - result = self.runner.invoke(ctl, ['pause', 'dummy']) - assert 'Success' in result.output - with patch('requests.patch', Mock(return_value=MockResponse(500))): result = self.runner.invoke(ctl, ['pause', 'dummy']) assert 'Failed' in result.output @@ -423,6 +420,17 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['pause', 'dummy']) assert 'Cluster is already paused' in result.output + with patch('requests.patch', Mock(return_value=MockResponse(200))): + result = self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + assert "'pause' request sent" in result.output + mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), + get_cluster(None, None, [], None, None)]) + self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + member = Member(1, 'other', 28, {}) + mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), + get_cluster(None, None, [member], None, None)]) + self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + @patch('patroni.ctl.get_dcs') def test_resume_cluster(self, mock_get_dcs): mock_get_dcs.return_value = self.e diff --git a/tests/test_ha.py b/tests/test_ha.py index ad513008..b037a41a 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -39,6 +39,7 @@ def get_cluster_initialized_without_leader(leader=False, failover=None, sync=Non m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', 'api_url': 'http://127.0.0.1:8011/patroni', 'state': 'running', + 'pause': True, 'tags': {'clonefrom': True}, 'scheduled_restart': {'schedule': "2100-01-01 10:53:07.560445+00:00", 'postgres_version': '99.0.0'}})