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

@@ -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):

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

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

tests/test_api.py

@@ -83,6 +83,10 @@ class MockHa(object):
     def wakeup():
         pass
 
+    @staticmethod
+    def is_paused():
+        return True
+
 
 class MockPatroni(object):
 

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

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:[email protected]: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'}})