From 296f4a7ff86bc8427dfa84e884e32f3a9c14e814 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Thu, 18 Aug 2016 17:40:16 +0200 Subject: [PATCH 01/30] Introduce disable/resume commands --- patroni/ctl.py | 46 +++++++++++++++++++++++++++++++++++++++++++++- patroni/ha.py | 2 ++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index b272d5ca..18301618 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -747,8 +747,13 @@ def touch_member(config, dcs): return dcs.touch_member(json.dumps(data, separators=(',', ':')), permanent=True) +def is_paused(cluster): + """Check if cluster management is paused""" + return 'pause' in cluster.config.data and cluster.config.data['pause'] + + def set_defaults(config, cluster_name): - ''' fill-in some basic configuration parameters if config file is not set ''' + """fill-in some basic configuration parameters if config file is not set """ config['postgresql'].setdefault('name', cluster_name) config['postgresql'].setdefault('scope', cluster_name) config['postgresql'].setdefault('listen', '127.0.0.1') @@ -800,3 +805,42 @@ def flush(cluster_name, member_names, config_file, dcs, force, role, target): check_response(r, member.name, 'flush scheduled restart') else: click.echo('No scheduled restart for member {0}'.format(member.name)) + + +@ctl.command('disable', help='Disable auto failover') +@click.argument('cluster_name') +@option_config_file +@option_dcs +def disable(config_file, cluster_name, dcs): + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + + if is_paused(cluster): + raise PatroniCtlException("Cluster is already paused") + + r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': True}, auth_header(config)) + if r.status_code == 200: + click.echo('Success: cluster management is paused.') + else: + click.echo('Failed: pause cluster management status code={0}, ({1})'.format(r.status_code, r.text)) + + return + + +@ctl.command('resume', help='Resume auto failover') +@click.argument('cluster_name') +@option_config_file +@option_dcs +def resume(config_file, cluster_name, dcs): + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + + if not is_paused(cluster): + raise PatroniCtlException("Cluster is not paused") + + r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': False}, auth_header(config)) + if r.status_code == 200: + click.echo('Success: cluster management is resumed') + else: + click.echo('Failed: resume cluster management, status code={0}, ({1})'.format(r.status_code, r.text)) + + return + diff --git a/patroni/ha.py b/patroni/ha.py index 3367572f..de8f976e 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -150,6 +150,8 @@ class Ha(object): if self.state_handler.is_leader() or self.state_handler.role == 'master': return message else: + if 'pause' in self.cluster.config.data and self.cluster.config.data['pause']: + return "Not promoted due to paused state." self.state_handler.promote() self.touch_member() return promote_message From b5d6b7d13d9b50cac2f35d3f44fc62bf921a868c Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Thu, 18 Aug 2016 17:53:18 +0200 Subject: [PATCH 02/30] Check if config is loaded --- patroni/ha.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/ha.py b/patroni/ha.py index de8f976e..0f0ef24d 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -150,7 +150,7 @@ class Ha(object): if self.state_handler.is_leader() or self.state_handler.role == 'master': return message else: - if 'pause' in self.cluster.config.data and self.cluster.config.data['pause']: + if self.cluster.config and 'pause' in self.cluster.config.data and self.cluster.config.data['pause']: return "Not promoted due to paused state." self.state_handler.promote() self.touch_member() From c50f072b314b1066c93519a241f6c8b8c917f8aa Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Tue, 23 Aug 2016 11:38:30 +0200 Subject: [PATCH 03/30] Avoid stopping pg instance when in paused state --- patroni/__init__.py | 5 ++++- patroni/ha.py | 5 +++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 37a8e848..ad968a08 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -128,5 +128,8 @@ def main(): pass finally: patroni.api.shutdown() - patroni.postgresql.stop(checkpoint=False) + if patroni.ha.is_paused(): + logger.info('Postgres is not stopped due paused state') + else: + patroni.postgresql.stop(checkpoint=False) patroni.dcs.delete_leader() diff --git a/patroni/ha.py b/patroni/ha.py index 0f0ef24d..28f7b696 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -26,6 +26,9 @@ class Ha(object): self.recovering = False self._async_executor = AsyncExecutor() + def is_paused(self): + return self.cluster.config and 'pause' in self.cluster.config.data and self.cluster.config.data['pause'] + def load_cluster_from_dcs(self): cluster = self.dcs.get_cluster() @@ -150,8 +153,6 @@ class Ha(object): if self.state_handler.is_leader() or self.state_handler.role == 'master': return message else: - if self.cluster.config and 'pause' in self.cluster.config.data and self.cluster.config.data['pause']: - return "Not promoted due to paused state." self.state_handler.promote() self.touch_member() return promote_message From 97f7576fab3c4d25d6cc40d556a44c0c5fac1fcc Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Tue, 23 Aug 2016 17:30:24 +0200 Subject: [PATCH 04/30] Do not drop active replication slots --- patroni/ha.py | 3 +++ patroni/postgresql.py | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/patroni/ha.py b/patroni/ha.py index 28f7b696..fa873977 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -346,6 +346,9 @@ class Ha(object): self.dcs.manual_failover('', '', index=self.cluster.failover.index) def process_unhealthy_cluster(self): + if self.is_paused(): + return "No action due to paused state" + if self.is_healthiest_node(): if self.acquire_lock(): if self.cluster.failover: diff --git a/patroni/postgresql.py b/patroni/postgresql.py index df2279f1..31f45148 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -900,7 +900,7 @@ $$""".format(name, ' '.join(options)), name, password, password) for slot in set(self._replication_slots) - set(slots): self.query("""SELECT pg_drop_replication_slot(%s) WHERE EXISTS(SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + WHERE slot_name = %s AND NOT active)""", slot, slot) # create new slots for slot in set(slots) - set(self._replication_slots): From a388fdb99db7df5c28c5b3cbeeb510358c72f853 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Wed, 24 Aug 2016 16:13:51 +0200 Subject: [PATCH 05/30] add paused state actions --- patroni/__init__.py | 4 ++-- patroni/ha.py | 28 ++++++++++++++++++++++++---- 2 files changed, 26 insertions(+), 6 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index ad968a08..6eba3790 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -129,7 +129,7 @@ def main(): finally: patroni.api.shutdown() if patroni.ha.is_paused(): - logger.info('Postgres is not stopped due paused state') + logger.info('Leader key is not deleted and Postgresql is not stopped due paused state') else: patroni.postgresql.stop(checkpoint=False) - patroni.dcs.delete_leader() + patroni.dcs.delete_leader() diff --git a/patroni/ha.py b/patroni/ha.py index fa873977..2ac2bc95 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -27,7 +27,8 @@ class Ha(object): self._async_executor = AsyncExecutor() def is_paused(self): - return self.cluster.config and 'pause' in self.cluster.config.data and self.cluster.config.data['pause'] + return self.cluster and self.cluster.config and 'pause' in self.cluster.config.data \ + and self.cluster.config.data['pause'] def load_cluster_from_dcs(self): cluster = self.dcs.get_cluster() @@ -346,9 +347,7 @@ class Ha(object): self.dcs.manual_failover('', '', index=self.cluster.failover.index) def process_unhealthy_cluster(self): - if self.is_paused(): - return "No action due to paused state" - + """Cluster has no leader key""" if self.is_healthiest_node(): if self.acquire_lock(): if self.cluster.failover: @@ -534,6 +533,24 @@ class Ha(object): return 'failed to start postgres' return None + def pause_action(self): + if not self.state_handler.is_healthy(): + return "Postgresql is not running" + + if not (self.state_handler.is_leader() or self.state_handler.role == 'master'): + return "I'm secondary" + + if self.has_lock(): + if not self.update_lock(): + # Either there is no connection to DCS or someone else acquired the lock + logger.error('failed to update leader lock') + self.load_cluster_from_dcs() + else: + if not self.acquire_lock(): + raise Exception("Someone already acquired the lock") + + return "I'm the leader" + def _run_cycle(self): try: self.load_cluster_from_dcs() @@ -550,6 +567,9 @@ class Ha(object): if self._async_executor.busy: return self.handle_long_action_in_progress() + if self.is_paused(): + return self.pause_action() + ". No action due to paused state" + # we've got here, so any async action has finished. Check if we tried to recover and failed if self.recovering: self.recovering = False From 4e61ef06a85cc89df1526c301957a87b4aaa0a13 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Wed, 24 Aug 2016 18:08:23 +0200 Subject: [PATCH 06/30] Add coverage in requirements Add some tests for patroni ctl --- patroni/ctl.py | 2 +- requirements.txt | 1 + tests/test_ctl.py | 37 +++++++++++++++++++++++++++++++++++++ 3 files changed, 39 insertions(+), 1 deletion(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 18301618..22fad384 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -749,7 +749,7 @@ def touch_member(config, dcs): def is_paused(cluster): """Check if cluster management is paused""" - return 'pause' in cluster.config.data and cluster.config.data['pause'] + return cluster.config and 'pause' in cluster.config.data and cluster.config.data['pause'] def set_defaults(config, cluster_name): diff --git a/requirements.txt b/requirements.txt index 8b9b3c79..0a8be460 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,3 +10,4 @@ click>=4.1 prettytable>=0.7 tzlocal python-dateutil +coverage \ No newline at end of file diff --git a/tests/test_ctl.py b/tests/test_ctl.py index d4e6db66..81a2261e 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -393,3 +393,40 @@ class TestCtl(unittest.TestCase): 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 + + @patch('patroni.ctl.get_dcs') + def test_disable_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, ['disable', 'dummy']) + assert 'Success' in result.output + + with patch('requests.patch', Mock(return_value=MockResponse(500))): + result = self.runner.invoke(ctl, ['disable', 'dummy']) + assert 'Failed' in result.output + + with patch('requests.patch', Mock(return_value=MockResponse(200))),\ + patch('patroni.ctl.is_paused', Mock(return_value=True)): + result = self.runner.invoke(ctl, ['disable', 'dummy']) + assert 'Cluster is already paused' in result.output + + @patch('patroni.ctl.get_dcs') + def test_resume_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('patroni.ctl.is_paused', Mock(return_value=True)): + with patch('requests.patch', Mock(return_value=MockResponse(200))): + result = self.runner.invoke(ctl, ['resume', 'dummy']) + assert 'Success' in result.output + + with patch('requests.patch', Mock(return_value=MockResponse(500))): + result = self.runner.invoke(ctl, ['resume', 'dummy']) + assert 'Failed' in result.output + + with patch('requests.patch', Mock(return_value=MockResponse(200))),\ + patch('patroni.ctl.is_paused', Mock(return_value=False)): + result = self.runner.invoke(ctl, ['resume', 'dummy']) + assert 'Cluster is not paused' in result.output From 5c63c9ffbd6174bce2290dc13694d191cf33bff7 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Thu, 25 Aug 2016 12:01:05 +0200 Subject: [PATCH 07/30] Pause state improvements --- patroni/ha.py | 18 ++++++++++-------- requirements.txt | 3 +-- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 2ac2bc95..3e4b3ba3 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -535,21 +535,23 @@ class Ha(object): def pause_action(self): if not self.state_handler.is_healthy(): - return "Postgresql is not running" + return "Postgresql is not running. No action due to paused state" - if not (self.state_handler.is_leader() or self.state_handler.role == 'master'): - return "I'm secondary" + if not self.state_handler.is_leader(): + if self.has_lock(): + self.dcs.delete_leader() + return "I'm secondary. No action due to paused state" if self.has_lock(): if not self.update_lock(): # Either there is no connection to DCS or someone else acquired the lock logger.error('failed to update leader lock') self.load_cluster_from_dcs() - else: + return "I'm the leader. Updating leader key" + elif self.cluster.is_unlocked(): if not self.acquire_lock(): - raise Exception("Someone already acquired the lock") - - return "I'm the leader" + return "Can't acquire the lock. No action due to paused state" + return "Cluster has no leader. Acquiring leader key" def _run_cycle(self): try: @@ -568,7 +570,7 @@ class Ha(object): return self.handle_long_action_in_progress() if self.is_paused(): - return self.pause_action() + ". No action due to paused state" + return self.pause_action() # we've got here, so any async action has finished. Check if we tried to recover and failed if self.recovering: diff --git a/requirements.txt b/requirements.txt index 0a8be460..59e00f8e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,5 +9,4 @@ python-consul==0.6.0 click>=4.1 prettytable>=0.7 tzlocal -python-dateutil -coverage \ No newline at end of file +python-dateutil \ No newline at end of file From 3977626fc24680a903164cba6b8069d62d2f9c4c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 26 Aug 2016 10:50:34 +0200 Subject: [PATCH 08/30] Bugfix: and has precedence over or --- patroni/api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/api.py b/patroni/api.py index 6f3a6ae9..9367496e 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -285,7 +285,7 @@ class RestApiHandler(BaseHTTPRequestHandler): return 503, 'Failover status unknown' def is_failover_possible(self, cluster, leader, candidate): - if leader and not cluster.leader or cluster.leader.name != leader: + if leader and (not cluster.leader or cluster.leader.name != leader): return 'leader name does not match' if candidate: members = [m for m in cluster.members if m.name == candidate] From 93b9046aed6b66f33f604f5c78d0146e36825bc6 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 26 Aug 2016 10:51:03 +0200 Subject: [PATCH 09/30] pep8 formatting --- patroni/ctl.py | 1 - 1 file changed, 1 deletion(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 22fad384..07ef8a64 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -843,4 +843,3 @@ def resume(config_file, cluster_name, dcs): click.echo('Failed: resume cluster management, status code={0}, ({1})'.format(r.status_code, r.text)) return - From ac49835a3c3df8df0a327a93225a509540aa910a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 26 Aug 2016 10:51:43 +0200 Subject: [PATCH 10/30] Possibility to disable automatic failover cluster-wide Any node of the cluster will maintain it's member key until Patroni is running there. Master node will also maintain the leader key until postgres is running as a master. If there is not postgres or it is running 'in_recovery', Patroni will release leader lock. Bootstrap of a new cluster will work (it is possible to specify paused: true) in the `bootstrap.dcs`. Replicas also will be able to join the cluster if the leader lock exist. If the postgres is not running on the node it will not try to bring it up. Also it disables reinitialize and all kind of scheduled actions, i.e. scheduled restart and scheduled failover. In case if DCS stops being reachable Patroni will not "demote" master if the automatic failover was disabled. Patroni will not stop postgres on exit. --- patroni/ha.py | 98 ++++++++++++++++++++++++++++-------------------- tests/test_ha.py | 48 ++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 40 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 3e4b3ba3..98982d12 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -27,8 +27,7 @@ class Ha(object): self._async_executor = AsyncExecutor() def is_paused(self): - return self.cluster and self.cluster.config and 'pause' in self.cluster.config.data \ - and self.cluster.config.data['pause'] + return self.cluster and self.cluster.config and self.cluster.config.data.get('pause', False) def load_cluster_from_dcs(self): cluster = self.dcs.get_cluster() @@ -142,10 +141,17 @@ class Ha(object): if recovery: ret = demote_reason if self.has_lock() else follow_reason else: - ret = demote_reason if self.state_handler.is_leader() else follow_reason + is_leader = self.state_handler.is_leader() + ret = demote_reason if is_leader else follow_reason node_to_follow = self._get_node_to_follow(self.cluster) + if self.is_paused(): + if is_leader: + return 'continue to run as master without lock' + elif not node_to_follow: + return 'no action' + self.state_handler.follow(node_to_follow, self.cluster.leader, recovery, self._async_executor) return ret @@ -227,6 +233,8 @@ class Ha(object): if failover.candidate: # manual failover to specific member if failover.candidate == self.state_handler.name: # manual failover to me return True + elif self.is_paused(): + return False # find specific node and check that it is healthy member = self.cluster.get_member(failover.candidate, fallback_to_leader=False) @@ -243,6 +251,8 @@ class Ha(object): # at this point we should consider all members as a candidates for failover # i.e. we assume that failover.candidate is None + elif self.is_paused(): + return False # try to pick some other members to failover and check that they are healthy if failover.leader: @@ -261,9 +271,15 @@ class Ha(object): return self._is_healthiest_node(members, check_replication_lag=False) def is_healthiest_node(self): + if self.is_paused() and self.cluster.failover and not self.cluster.failover.scheduled_at: + return self.manual_failover_process_no_leader() + if self.state_handler.is_leader(): # leader is always the healthiest return True + if self.is_paused(): + return False + if self.patroni.nofailover: # nofailover tag makes node always unhealthy return False @@ -323,28 +339,35 @@ class Ha(object): def process_manual_failover_from_leader(self): failover = self.cluster.failover + if failover.scheduled_at and self.is_paused(): + return + if (failover.scheduled_at and not self.should_run_scheduled_action("failover", failover.scheduled_at, lambda: - self.dcs.manual_failover('', '', index=self.cluster.failover.index))): + self.dcs.manual_failover('', '', index=failover.index))): return if not failover.leader or failover.leader == self.state_handler.name: if not failover.candidate or failover.candidate != self.state_handler.name: - members = [m for m in self.cluster.members if not failover.candidate or m.name == failover.candidate] - if self.is_failover_possible(members): # check that there are healthy members - self._async_executor.schedule('manual failover: demote') - self._async_executor.run_async(self.demote) - return 'manual failover: demoting myself' + if not failover.candidate and self.is_paused(): + logger.warning('Failover is possible only to a specific candidate in a paused state') else: - logger.warning('manual failover: no healthy members found, failover is not possible') + members = [m for m in self.cluster.members + if not failover.candidate or m.name == failover.candidate] + if self.is_failover_possible(members): # check that there are healthy members + self._async_executor.schedule('manual failover: demote') + self._async_executor.run_async(self.demote) + return 'manual failover: demoting myself' + else: + logger.warning('manual failover: no healthy members found, failover is not possible') else: logger.warning('manual failover: I am already the leader, no need to failover') else: logger.warning('manual failover: leader name does not match: %s != %s', - self.cluster.failover.leader, self.state_handler.name) + failover.leader, self.state_handler.name) logger.info('Trying to clean up failover key') - self.dcs.manual_failover('', '', index=self.cluster.failover.index) + self.dcs.manual_failover('', '', index=failover.index) def process_unhealthy_cluster(self): """Cluster has no leader key""" @@ -363,7 +386,7 @@ class Ha(object): if self.patroni.nofailover: return self.follow('demoting self because I am not allowed to become master', 'following a different leader because I am not allowed to promote') - return self.follow('demoting self because i am not the healthiest node', + return self.follow('demoting self because i am not the healthiest node', # should not happen in real life 'following a different leader because i am not the healthiest node') def process_healthy_cluster(self): @@ -373,6 +396,11 @@ class Ha(object): if msg is not None: return msg + if self.is_paused() and not self.state_handler.is_leader(): + self.dcs.delete_leader() + self.dcs.reset_cluster() + return 'removed leader lock because postgres is not running as master' + if self.update_lock(): return self.enforce_master_role('no action. i am the leader with the lock', 'promoted self to leader because i had the session lock') @@ -386,6 +414,8 @@ class Ha(object): 'no action. i am a secondary and i am following a leader', False) def evaluate_scheduled_restart(self): + if self.is_paused(): + return None # restart if we need to restart_data = self.future_restart_scheduled() if restart_data: @@ -496,7 +526,10 @@ class Ha(object): def process_scheduled_action(self): if self.reinitialize_scheduled(): - if self.cluster.is_unlocked(): + if self.is_paused(): + logger.warning('Cluster is in a pause state, can not reinitialize') + self._async_executor.reset_scheduled_action() + elif self.cluster.is_unlocked(): logger.error('Cluster has no leader, can not reinitialize') self._async_executor.reset_scheduled_action() elif self.has_lock(): @@ -533,26 +566,6 @@ class Ha(object): return 'failed to start postgres' return None - def pause_action(self): - if not self.state_handler.is_healthy(): - return "Postgresql is not running. No action due to paused state" - - if not self.state_handler.is_leader(): - if self.has_lock(): - self.dcs.delete_leader() - return "I'm secondary. No action due to paused state" - - if self.has_lock(): - if not self.update_lock(): - # Either there is no connection to DCS or someone else acquired the lock - logger.error('failed to update leader lock') - self.load_cluster_from_dcs() - return "I'm the leader. Updating leader key" - elif self.cluster.is_unlocked(): - if not self.acquire_lock(): - return "Can't acquire the lock. No action due to paused state" - return "Cluster has no leader. Acquiring leader key" - def _run_cycle(self): try: self.load_cluster_from_dcs() @@ -569,9 +582,6 @@ class Ha(object): if self._async_executor.busy: return self.handle_long_action_in_progress() - if self.is_paused(): - return self.pause_action() - # we've got here, so any async action has finished. Check if we tried to recover and failed if self.recovering: self.recovering = False @@ -588,7 +598,7 @@ class Ha(object): if self.state_handler.data_directory_empty(): return self.bootstrap() # new node # "bootstrap", but data directory is not empty - elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked(): + elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked() and not self.is_paused(): self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid) else: # check if we are allowed to join @@ -599,6 +609,13 @@ class Ha(object): # try to start dead postgres if not self.state_handler.is_healthy(): + if self.is_paused(): + if self.has_lock(): + self.dcs.delete_leader() + self.dcs.reset_cluster() + return 'removed leader lock because postgres is not running' + else: + return 'postgres is not running' msg = self.recover() if msg is not None: return msg @@ -621,12 +638,13 @@ class Ha(object): self.state_handler.sync_replication_slots(self.cluster) except DCSError: logger.error('Error communicating with DCS') - if self.state_handler.is_running() and self.state_handler.is_leader(): + if not self.is_paused() and self.state_handler.is_running() and self.state_handler.is_leader(): self.demote(delete_leader=False) return 'demoted self because DCS is not accessible and i was a leader' + return 'DCS is not accessible' except (psycopg2.Error, PostgresConnectionException): logger.exception('Error communicating with PostgreSQL. Will try again later') def run_cycle(self): with self._async_executor: - return self._run_cycle() + return (self.is_paused() and 'PAUSE: ' or '') + self._run_cycle() diff --git a/tests/test_ha.py b/tests/test_ha.py index de458b60..0b5ed96a 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -166,6 +166,9 @@ class TestHa(unittest.TestCase): self.ha.cluster = get_cluster_initialized_with_leader() self.assertEquals(self.ha.run_cycle(), 'starting as readonly because i had the session lock') + def test_do_not_recover_in_pause(self): + pass + @patch('sys.exit', return_value=1) @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True)) def test_sysid_no_match(self, exit_mock): @@ -234,6 +237,13 @@ class TestHa(unittest.TestCase): self.ha.patroni.replicatefrom = "foo" self.assertEquals(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') + def test_follow_in_pause(self): + self.ha.cluster.is_unlocked = false + self.ha.is_paused = true + self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock') + self.p.is_leader = false + self.assertEquals(self.ha.run_cycle(), 'PAUSE: no action') + def test_no_etcd_connection_master_demote(self): self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) self.assertEquals(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader') @@ -272,6 +282,11 @@ class TestHa(unittest.TestCase): self.ha.run_cycle() self.assertIsNone(self.ha._async_executor.scheduled_action) + with patch.object(Ha, 'is_paused', true): + self.ha.schedule_reinitialize() + self.ha.run_cycle() + self.assertIsNone(self.ha._async_executor.scheduled_action) + self.ha.cluster = get_cluster_initialized_with_leader() self.ha.has_lock = true self.ha.schedule_reinitialize() @@ -345,6 +360,16 @@ class TestHa(unittest.TestCase): self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.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_from_leader_in_pause(self): + self.ha.has_lock = true + self.ha.is_paused = true + scheduled = datetime.datetime.now() + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled)) + self.assertEquals('PAUSE: no action. i am the leader with the lock', self.ha.run_cycle()) + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None)) + self.assertEquals('PAUSE: 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 @@ -370,11 +395,20 @@ class TestHa(unittest.TestCase): self.ha.patroni.nofailover = True self.assertEquals(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote') + def test_manual_failover_process_no_leader_in_pause(self): + self.ha.is_paused = true + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None)) + self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock') + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', '', None)) + self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock') + def test_is_healthiest_node(self): self.ha.state_handler.is_leader = false self.ha.patroni.nofailover = False self.ha.fetch_node_status = lambda e: (e, True, True, 0, {}) self.assertTrue(self.ha.is_healthiest_node()) + self.ha.is_paused = true + self.assertFalse(self.ha.is_healthiest_node()) def test__is_healthiest_node(self): self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) @@ -451,3 +485,17 @@ class TestHa(unittest.TestCase): self.p._pending_restart = False self.assertFalse(self.ha.restart_matches("replica", "9.5.2", True)) self.assertTrue(self.ha.restart_matches("replica", "9.5.2", False)) + + def test_process_healthy_cluster_in_pause(self): + self.p.is_leader = false + self.ha.is_paused = true + self.p.name = 'leader' + self.ha.cluster = get_cluster_initialized_with_leader() + self.assertEquals(self.ha.run_cycle(), 'PAUSE: removed leader lock because postgres is not running as master') + + def test_postgres_unhealthy_in_pause(self): + self.ha.is_paused = true + self.p.is_healthy = false + self.assertEquals(self.ha.run_cycle(), 'PAUSE: postgres is not running') + self.ha.has_lock = true + self.assertEquals(self.ha.run_cycle(), 'PAUSE: removed leader lock because postgres is not running') From 89ef5da5aee9c1ae2b58795fc9ea417edbb33c73 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Mon, 29 Aug 2016 08:36:35 +0200 Subject: [PATCH 11/30] Add tests for api; add checks for ctl and api for the paused state case --- patroni/api.py | 29 +++++++++++++++++++++++------ patroni/ctl.py | 17 ++++++++++++----- tests/test_api.py | 17 +++++++++++++++++ tests/test_ctl.py | 11 +++++++++++ tests/test_ha.py | 5 +++++ 5 files changed, 68 insertions(+), 11 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 9367496e..71257b68 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -199,6 +199,11 @@ class RestApiHandler(BaseHTTPRequestHandler): return if request: logger.debug("received restart request: {0}".format(request)) + + if self.server.patroni.ha.is_paused() and 'schedule' in request and self: + self._write_response(status_code, "Can't schedule restart in the paused state") + return + for k in request: if k == 'schedule': (_, data, request[k]) = self.parse_schedule(request[k], "restart") @@ -239,18 +244,27 @@ class RestApiHandler(BaseHTTPRequestHandler): @check_auth def do_DELETE_restart(self): - if self.server.patroni.ha.delete_future_restart(): - data = "scheduled restart deleted" - code = 200 + if self.server.patroni.ha.is_paused(): + data = "Can't delete scheduled restart in the paused state" + code = 500 else: - data = "no restarts are scheduled" - code = 404 + if self.server.patroni.ha.delete_future_restart(): + data = "scheduled restart deleted" + code = 200 + else: + data = "no restarts are scheduled" + code = 404 self._write_response(code, data) @check_auth def do_POST_reinitialize(self): patroni = self.server.patroni cluster = patroni.dcs.get_cluster() + status_code = 500 + if self.server.patroni.ha.is_paused(): + self._write_response(status_code, "Can't do reinitialize in the paused state") + return + if cluster.is_unlocked(): status_code = 503 data = 'Cluster has no leader, can not reinitialize' @@ -303,14 +317,17 @@ class RestApiHandler(BaseHTTPRequestHandler): @check_auth def do_POST_failover(self): request = self._read_json_content() + status_code = 500 if not request: return leader = request.get('leader') candidate = request.get('candidate') or request.get('member') scheduled_at = request.get('scheduled_at') + if scheduled_at and self.server.patroni.ha.is_paused(): + self._write_response(status_code, "Can't schedule failover in the paused state") + cluster = self.server.patroni.dcs.get_cluster() - status_code = 500 logger.info("received failover request with leader=%s candidate=%s scheduled_at=%s", leader, candidate, scheduled_at) diff --git a/patroni/ctl.py b/patroni/ctl.py index 07ef8a64..b5dbbc28 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -499,6 +499,8 @@ def restart(cluster_name, member_names, config_file, dcs, force, role, p_any, sc scheduled_at = parse_scheduled(scheduled) if scheduled_at: + if is_paused(cluster): + raise PatroniCtlException("Can't schedule restart in the paused state") content['schedule'] = scheduled_at.isoformat() for member in members: @@ -554,17 +556,17 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) - if cluster.leader is None: + if cluster.leader is None and not is_paused(cluster): raise PatroniCtlException('This cluster has no master') - if master is None: + if master is None and (not is_paused(cluster) or cluster.leader): if force: master = cluster.leader.member.name else: master = click.prompt('Master', type=str, default=cluster.leader.member.name) - if cluster.leader.member.name != master: - raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.format(master, cluster_name)) + if not is_paused(cluster) and cluster.leader.member.name != master: + raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.format(master, cluster_name)) candidate_names = [str(m.name) for m in cluster.members if m.name != master] # We sort the names for consistent output to the client @@ -589,9 +591,14 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled scheduled_at = parse_scheduled(scheduled) if scheduled_at: + if is_paused(cluster): + raise PatroniCtlException("Can't schedule failover in the paused state") scheduled_at = scheduled_at.isoformat() - failover_value = {'leader': master, 'candidate': candidate, 'scheduled_at': scheduled_at} + failover_value = {'candidate': candidate, 'scheduled_at': scheduled_at} + if master: + failover_value['leader'] = master + logging.debug(failover_value) # By now we have established that the leader exists and the candidate exists diff --git a/tests/test_api.py b/tests/test_api.py index 3e90346a..cc5f590e 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -39,6 +39,10 @@ class MockHa(object): state_handler = MockPostgresql() + @staticmethod + def is_paused(): + return False + @staticmethod def schedule_reinitialize(): return 'reinitialize' @@ -221,11 +225,18 @@ class TestRestApiHandler(unittest.TestCase): request = make_request(role='master', postgres_version='9.5.2') MockRestApiServer(RestApiHandler, request) + with patch.object(MockHa, 'is_paused', Mock(return_value=True)): + request = make_request(schedule='2016-08-42 12:45TZ+1', role='master') + MockRestApiServer(RestApiHandler, request) + def test_do_DELETE_restart(self): for retval in (True, False): with patch.object(MockHa, 'delete_future_restart', Mock(return_value=retval)): request = 'DELETE /restart HTTP/1.0' + self._authorization self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) + with patch.object(MockHa, 'is_paused', Mock(return_value=True)): + request = 'DELETE /restart HTTP/1.0' + self._authorization + self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) @patch.object(MockPatroni, 'dcs') def test_do_POST_reinitialize(self, dcs): @@ -239,6 +250,9 @@ class TestRestApiHandler(unittest.TestCase): cluster.leader.name = 'test' self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) + with patch.object(MockHa, 'is_paused', Mock(return_value=True)): + self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) + @patch('time.sleep', Mock()) def test_RestApiServer_query(self): with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)): @@ -306,6 +320,9 @@ class TestRestApiHandler(unittest.TestCase): d.manual_failover.return_value = False MockRestApiServer(RestApiHandler, request) + with patch.object(MockHa, 'is_paused', Mock(return_value=True)): + MockRestApiServer(RestApiHandler, request) + # Exception: No timezone specified request = post + '97\n\n{"leader": "postgresql1", "member": "postgresql2",' +\ ' "scheduled_at": "6016-02-15T18:13:30.568224"}' diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 81a2261e..e9a1bc82 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -82,6 +82,11 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n2030-01-01T12:23:00\ny') assert result.exit_code == 0 + with patch('patroni.ctl.is_paused', Mock(return_value=True)): + result = self.runner.invoke(ctl, + ['failover', 'dummy', '--force', '--scheduled', '2015-01-01T12:00:00+01:00']) + assert result.exit_code == 1 + # Aborting failover,as we anser NO to the confirmation result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n\nN') assert result.exit_code == 1 @@ -241,6 +246,12 @@ class TestCtl(unittest.TestCase): '--scheduled', '2300-10-01T14:30']) assert 'Failed: flush scheduled restart' in result.output + with patch('patroni.ctl.is_paused', Mock(return_value=True)): + result = self.runner.invoke(ctl, + ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30']) + assert result.exit_code == 1 + + with patch('requests.post', Mock(return_value=MockResponse())): # normal restart, the schedule is actually parsed, but not validated in patronictl result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0.0', diff --git a/tests/test_ha.py b/tests/test_ha.py index 0b5ed96a..69293cec 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -499,3 +499,8 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'PAUSE: postgres is not running') self.ha.has_lock = true self.assertEquals(self.ha.run_cycle(), 'PAUSE: removed leader lock because postgres is not running') + + def test_no_etcd_connection_in_pause(self): + self.ha.is_paused = true + self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) + self.assertEquals(self.ha.run_cycle(), 'PAUSE: DCS is not accessible') From 3d1fe3fa49ede4c4e88ae11353661157361768c6 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Mon, 29 Aug 2016 09:29:49 +0200 Subject: [PATCH 12/30] Introduce is_paused method in the Cluster --- patroni/api.py | 25 +++++++++++-------------- patroni/ctl.py | 25 +++++++++---------------- patroni/dcs/__init__.py | 3 +++ patroni/ha.py | 2 +- tests/test_api.py | 3 --- tests/test_ctl.py | 11 +++++------ 6 files changed, 29 insertions(+), 40 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 71257b68..7e066289 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -194,13 +194,14 @@ class RestApiHandler(BaseHTTPRequestHandler): status_code = 500 data = 'restart failed' request = self._read_json_content(body_is_optional=True) + cluster = self.server.patroni.dcs.get_cluster() if request is None: # failed to parse the json return if request: logger.debug("received restart request: {0}".format(request)) - if self.server.patroni.ha.is_paused() and 'schedule' in request and self: + if cluster.is_paused() and 'schedule' in request: self._write_response(status_code, "Can't schedule restart in the paused state") return @@ -244,16 +245,12 @@ class RestApiHandler(BaseHTTPRequestHandler): @check_auth def do_DELETE_restart(self): - if self.server.patroni.ha.is_paused(): - data = "Can't delete scheduled restart in the paused state" - code = 500 + if self.server.patroni.ha.delete_future_restart(): + data = "scheduled restart deleted" + code = 200 else: - if self.server.patroni.ha.delete_future_restart(): - data = "scheduled restart deleted" - code = 200 - else: - data = "no restarts are scheduled" - code = 404 + data = "no restarts are scheduled" + code = 404 self._write_response(code, data) @check_auth @@ -261,7 +258,7 @@ class RestApiHandler(BaseHTTPRequestHandler): patroni = self.server.patroni cluster = patroni.dcs.get_cluster() status_code = 500 - if self.server.patroni.ha.is_paused(): + if cluster.is_paused(): self._write_response(status_code, "Can't do reinitialize in the paused state") return @@ -324,11 +321,11 @@ class RestApiHandler(BaseHTTPRequestHandler): leader = request.get('leader') candidate = request.get('candidate') or request.get('member') scheduled_at = request.get('scheduled_at') - if scheduled_at and self.server.patroni.ha.is_paused(): - self._write_response(status_code, "Can't schedule failover in the paused state") - cluster = self.server.patroni.dcs.get_cluster() + if scheduled_at and cluster.is_paused(): + self._write_response(status_code, "Can't schedule failover in the paused state") + logger.info("received failover request with leader=%s candidate=%s scheduled_at=%s", leader, candidate, scheduled_at) diff --git a/patroni/ctl.py b/patroni/ctl.py index b5dbbc28..3ad22e2a 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -499,7 +499,7 @@ def restart(cluster_name, member_names, config_file, dcs, force, role, p_any, sc scheduled_at = parse_scheduled(scheduled) if scheduled_at: - if is_paused(cluster): + if cluster.is_paused(): raise PatroniCtlException("Can't schedule restart in the paused state") content['schedule'] = scheduled_at.isoformat() @@ -556,17 +556,17 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) - if cluster.leader is None and not is_paused(cluster): + if cluster.leader is None and not cluster.is_paused(): raise PatroniCtlException('This cluster has no master') - if master is None and (not is_paused(cluster) or cluster.leader): + if master is None and (not cluster.is_paused() or cluster.leader): if force: master = cluster.leader.member.name else: master = click.prompt('Master', type=str, default=cluster.leader.member.name) - if not is_paused(cluster) and cluster.leader.member.name != master: - raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.format(master, cluster_name)) + if not (master is not None and cluster.leader and cluster.leader.member.name == master): + raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.format(master, cluster_name)) candidate_names = [str(m.name) for m in cluster.members if m.name != master] # We sort the names for consistent output to the client @@ -591,13 +591,11 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled scheduled_at = parse_scheduled(scheduled) if scheduled_at: - if is_paused(cluster): + if cluster.is_paused(): raise PatroniCtlException("Can't schedule failover in the paused state") scheduled_at = scheduled_at.isoformat() - failover_value = {'candidate': candidate, 'scheduled_at': scheduled_at} - if master: - failover_value['leader'] = master + failover_value = {'leader': master, 'candidate': candidate, 'scheduled_at': scheduled_at} logging.debug(failover_value) @@ -754,11 +752,6 @@ def touch_member(config, dcs): return dcs.touch_member(json.dumps(data, separators=(',', ':')), permanent=True) -def is_paused(cluster): - """Check if cluster management is paused""" - return cluster.config and 'pause' in cluster.config.data and cluster.config.data['pause'] - - def set_defaults(config, cluster_name): """fill-in some basic configuration parameters if config file is not set """ config['postgresql'].setdefault('name', cluster_name) @@ -821,7 +814,7 @@ def flush(cluster_name, member_names, config_file, dcs, force, role, target): def disable(config_file, cluster_name, dcs): config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) - if is_paused(cluster): + if cluster.is_paused(): raise PatroniCtlException("Cluster is already paused") r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': True}, auth_header(config)) @@ -840,7 +833,7 @@ def disable(config_file, cluster_name, dcs): def resume(config_file, cluster_name, dcs): config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) - if not is_paused(cluster): + if not cluster.is_paused(): raise PatroniCtlException("Cluster is not paused") r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': False}, auth_header(config)) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index d7a69997..bbeab3c5 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -228,6 +228,9 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat candidates = [m for m in self.members if m.clonefrom and (not self.leader or m.name != self.leader.name)] return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader + def is_paused(self): + return self.config and self.config.data.get('pause', False) + @six.add_metaclass(abc.ABCMeta) class AbstractDCS(object): diff --git a/patroni/ha.py b/patroni/ha.py index 5fb0ea6c..537b6e57 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -27,7 +27,7 @@ class Ha(object): self._async_executor = AsyncExecutor() def is_paused(self): - return self.cluster and self.cluster.config and self.cluster.config.data.get('pause', False) + return self.cluster and self.cluster.is_paused() def load_cluster_from_dcs(self): cluster = self.dcs.get_cluster() diff --git a/tests/test_api.py b/tests/test_api.py index cc5f590e..83aff049 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -234,9 +234,6 @@ class TestRestApiHandler(unittest.TestCase): with patch.object(MockHa, 'delete_future_restart', Mock(return_value=retval)): request = 'DELETE /restart HTTP/1.0' + self._authorization self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) - with patch.object(MockHa, 'is_paused', Mock(return_value=True)): - request = 'DELETE /restart HTTP/1.0' + self._authorization - self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) @patch.object(MockPatroni, 'dcs') def test_do_POST_reinitialize(self, dcs): diff --git a/tests/test_ctl.py b/tests/test_ctl.py index e9a1bc82..2237145e 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -82,7 +82,7 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n2030-01-01T12:23:00\ny') assert result.exit_code == 0 - with patch('patroni.ctl.is_paused', Mock(return_value=True)): + with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): result = self.runner.invoke(ctl, ['failover', 'dummy', '--force', '--scheduled', '2015-01-01T12:00:00+01:00']) assert result.exit_code == 1 @@ -246,12 +246,11 @@ class TestCtl(unittest.TestCase): '--scheduled', '2300-10-01T14:30']) assert 'Failed: flush scheduled restart' in result.output - with patch('patroni.ctl.is_paused', Mock(return_value=True)): + with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30']) assert result.exit_code == 1 - with patch('requests.post', Mock(return_value=MockResponse())): # normal restart, the schedule is actually parsed, but not validated in patronictl result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0.0', @@ -419,7 +418,7 @@ class TestCtl(unittest.TestCase): assert 'Failed' in result.output with patch('requests.patch', Mock(return_value=MockResponse(200))),\ - patch('patroni.ctl.is_paused', Mock(return_value=True)): + patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): result = self.runner.invoke(ctl, ['disable', 'dummy']) assert 'Cluster is already paused' in result.output @@ -428,7 +427,7 @@ class TestCtl(unittest.TestCase): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - with patch('patroni.ctl.is_paused', Mock(return_value=True)): + with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): with patch('requests.patch', Mock(return_value=MockResponse(200))): result = self.runner.invoke(ctl, ['resume', 'dummy']) assert 'Success' in result.output @@ -438,6 +437,6 @@ class TestCtl(unittest.TestCase): assert 'Failed' in result.output with patch('requests.patch', Mock(return_value=MockResponse(200))),\ - patch('patroni.ctl.is_paused', Mock(return_value=False)): + patch('patroni.dcs.Cluster.is_paused', Mock(return_value=False)): result = self.runner.invoke(ctl, ['resume', 'dummy']) assert 'Cluster is not paused' in result.output From 9fdd021e08cbc7f8a41a72e1ee1f518a485f2c35 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 29 Aug 2016 10:25:46 +0200 Subject: [PATCH 13/30] Fix unit-tests for api --- tests/test_api.py | 25 ++++++++++--------------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/tests/test_api.py b/tests/test_api.py index 83aff049..ad96ed87 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -39,10 +39,6 @@ class MockHa(object): state_handler = MockPostgresql() - @staticmethod - def is_paused(): - return False - @staticmethod def schedule_reinitialize(): return 'reinitialize' @@ -183,7 +179,9 @@ class TestRestApiHandler(unittest.TestCase): MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization) self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization)) - def test_do_POST_restart(self): + @patch.object(MockPatroni, 'dcs') + def test_do_POST_restart(self, mock_dcs): + mock_dcs.get_cluster.return_value.is_paused.return_value = False request = 'POST /restart HTTP/1.0' + self._authorization self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) @@ -225,9 +223,8 @@ class TestRestApiHandler(unittest.TestCase): request = make_request(role='master', postgres_version='9.5.2') MockRestApiServer(RestApiHandler, request) - with patch.object(MockHa, 'is_paused', Mock(return_value=True)): - request = make_request(schedule='2016-08-42 12:45TZ+1', role='master') - MockRestApiServer(RestApiHandler, request) + mock_dcs.get_cluster.return_value.is_paused.return_value = True + MockRestApiServer(RestApiHandler, make_request(schedule='2016-08-42 12:45TZ+1', role='master')) def test_do_DELETE_restart(self): for retval in (True, False): @@ -236,8 +233,9 @@ class TestRestApiHandler(unittest.TestCase): self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) @patch.object(MockPatroni, 'dcs') - def test_do_POST_reinitialize(self, dcs): - cluster = dcs.get_cluster.return_value + def test_do_POST_reinitialize(self, mock_dcs): + cluster = mock_dcs.get_cluster.return_value + cluster.is_paused.return_value = False request = 'POST /reinitialize HTTP/1.0' + self._authorization MockRestApiServer(RestApiHandler, request) cluster.is_unlocked.return_value = False @@ -247,8 +245,8 @@ class TestRestApiHandler(unittest.TestCase): cluster.leader.name = 'test' self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) - with patch.object(MockHa, 'is_paused', Mock(return_value=True)): - self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) + cluster.is_paused.return_value = True + self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) @patch('time.sleep', Mock()) def test_RestApiServer_query(self): @@ -317,9 +315,6 @@ class TestRestApiHandler(unittest.TestCase): d.manual_failover.return_value = False MockRestApiServer(RestApiHandler, request) - with patch.object(MockHa, 'is_paused', Mock(return_value=True)): - MockRestApiServer(RestApiHandler, request) - # Exception: No timezone specified request = post + '97\n\n{"leader": "postgresql1", "member": "postgresql2",' +\ ' "scheduled_at": "6016-02-15T18:13:30.568224"}' From e643321ab70dd5aab0a0ac16f8c6caa63b4507c0 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 29 Aug 2016 11:34:34 +0200 Subject: [PATCH 14/30] Address code-review --- patroni/ha.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 537b6e57..75be52a9 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -271,7 +271,8 @@ class Ha(object): return self._is_healthiest_node(members, check_replication_lag=False) def is_healthiest_node(self): - if self.is_paused() and self.cluster.failover and not self.cluster.failover.scheduled_at: + if self.is_paused() and not self.patroni.nofailover and \ + self.cluster.failover and not self.cluster.failover.scheduled_at: return self.manual_failover_process_no_leader() if self.state_handler.is_leader(): # leader is always the healthiest @@ -305,7 +306,7 @@ class Ha(object): self.state_handler.follow(None, None) def should_run_scheduled_action(self, action_name, scheduled_at, cleanup_fn): - if scheduled_at: + if scheduled_at and not self.is_paused(): # If the scheduled action is in the far future, we shouldn't do anything and just return. # If the scheduled action is in the past, we consider the value to be stale and we remove # the value. @@ -339,9 +340,6 @@ class Ha(object): def process_manual_failover_from_leader(self): failover = self.cluster.failover - if failover.scheduled_at and self.is_paused(): - return - if (failover.scheduled_at and not self.should_run_scheduled_action("failover", failover.scheduled_at, lambda: self.dcs.manual_failover('', '', index=failover.index))): @@ -414,8 +412,6 @@ class Ha(object): 'no action. i am a secondary and i am following a leader', False) def evaluate_scheduled_restart(self): - if self.is_paused(): - return None # restart if we need to restart_data = self.future_restart_scheduled() if restart_data: From 22e4af3fb1283de061d6e541db345fe8e4a12fd6 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Mon, 29 Aug 2016 12:04:30 +0200 Subject: [PATCH 15/30] Fix failover in the paused state --- patroni/ctl.py | 4 ++-- tests/test_ctl.py | 5 ----- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 3ad22e2a..d47239db 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -559,13 +559,13 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled if cluster.leader is None and not cluster.is_paused(): raise PatroniCtlException('This cluster has no master') - if master is None and (not cluster.is_paused() or cluster.leader): + if master is None and not cluster.is_paused(): if force: master = cluster.leader.member.name else: master = click.prompt('Master', type=str, default=cluster.leader.member.name) - if not (master is not None and cluster.leader and cluster.leader.member.name == master): + if master is None or not cluster.leader or cluster.leader.member.name != master: raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.format(master, cluster_name)) candidate_names = [str(m.name) for m in cluster.members if m.name != master] diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 2237145e..c8c93ea1 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -82,11 +82,6 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n2030-01-01T12:23:00\ny') assert result.exit_code == 0 - with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): - result = self.runner.invoke(ctl, - ['failover', 'dummy', '--force', '--scheduled', '2015-01-01T12:00:00+01:00']) - assert result.exit_code == 1 - # Aborting failover,as we anser NO to the confirmation result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n\nN') assert result.exit_code == 1 From 62f14dfd10243b6d18640c6d35c9387d28d1abc8 Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Mon, 29 Aug 2016 12:29:34 +0200 Subject: [PATCH 16/30] Fix master check --- patroni/ctl.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index d47239db..1e8feca3 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -565,7 +565,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled else: master = click.prompt('Master', type=str, default=cluster.leader.member.name) - if master is None or not cluster.leader or cluster.leader.member.name != master: + if master is not None and cluster.leader and cluster.leader.member.name != master: raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.format(master, cluster_name)) candidate_names = [str(m.name) for m in cluster.members if m.name != master] From 799d4c9bb8eb418a1f2169ce41208b8f6ad28abd Mon Sep 17 00:00:00 2001 From: Murat Kabilov Date: Mon, 29 Aug 2016 14:30:19 +0200 Subject: [PATCH 17/30] Disable command renamed to pause --- patroni/ctl.py | 50 +++++++++++++++++++++-------------------------- tests/test_ctl.py | 13 ++++++++---- 2 files changed, 31 insertions(+), 32 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 1e8feca3..4288089a 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -559,7 +559,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled if cluster.leader is None and not cluster.is_paused(): raise PatroniCtlException('This cluster has no master') - if master is None and not cluster.is_paused(): + if master is None and (not cluster.is_paused() or cluster.leader): if force: master = cluster.leader.member.name else: @@ -612,7 +612,9 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled r = None try: - r = request_patroni(cluster.leader.member, 'post', 'failover', failover_value, auth_header(config)) + member = cluster.leader.member if cluster.leader else [m for m in cluster.members if m.name == candidate][0] + + r = request_patroni(member, 'post', 'failover', failover_value, auth_header(config)) if r.status_code in (200, 202): logging.debug(r) cluster = dcs.get_cluster() @@ -807,23 +809,26 @@ def flush(cluster_name, member_names, config_file, dcs, force, role, target): click.echo('No scheduled restart for member {0}'.format(member.name)) -@ctl.command('disable', help='Disable auto failover') +def toggle_pause(config_file, cluster_name, dcs, paused): + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + if cluster.is_paused() == paused: + raise PatroniCtlException("Cluster " + ("is already" if paused else "is not") + " paused") + + r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': paused}, auth_header(config)) + + if r.status_code == 200: + click.echo("Success: cluster management is " + ("paused" if paused else "resumed")) + else: + click.echo("Failed: " + ("pause" if paused else "resume") + + " cluster management status code={0}, ({1})".format(r.status_code, r.text)) + + +@ctl.command('pause', help='Disable auto failover') @click.argument('cluster_name') @option_config_file @option_dcs -def disable(config_file, cluster_name, dcs): - config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) - - if cluster.is_paused(): - raise PatroniCtlException("Cluster is already paused") - - r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': True}, auth_header(config)) - if r.status_code == 200: - click.echo('Success: cluster management is paused.') - else: - click.echo('Failed: pause cluster management status code={0}, ({1})'.format(r.status_code, r.text)) - - return +def pause(config_file, cluster_name, dcs): + return toggle_pause(config_file, cluster_name, dcs, True) @ctl.command('resume', help='Resume auto failover') @@ -831,15 +836,4 @@ def disable(config_file, cluster_name, dcs): @option_config_file @option_dcs def resume(config_file, cluster_name, dcs): - config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) - - if not cluster.is_paused(): - raise PatroniCtlException("Cluster is not paused") - - r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': False}, auth_header(config)) - if r.status_code == 200: - click.echo('Success: cluster management is resumed') - else: - click.echo('Failed: resume cluster management, status code={0}, ({1})'.format(r.status_code, r.text)) - - return + return toggle_pause(config_file, cluster_name, dcs, False) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index c8c93ea1..dcdb7268 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -82,6 +82,11 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n2030-01-01T12:23:00\ny') assert result.exit_code == 0 + with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): + result = self.runner.invoke(ctl, + ['failover', 'dummy', '--force', '--scheduled', '2015-01-01T12:00:00+01:00']) + assert result.exit_code == 1 + # Aborting failover,as we anser NO to the confirmation result = self.runner.invoke(ctl, ['failover', 'dummy'], input='leader\nother\n\nN') assert result.exit_code == 1 @@ -400,21 +405,21 @@ class TestCtl(unittest.TestCase): assert 'Failed: flush scheduled restart' in result.output @patch('patroni.ctl.get_dcs') - def test_disable_cluster(self, mock_get_dcs): + 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, ['disable', 'dummy']) + 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, ['disable', 'dummy']) + result = self.runner.invoke(ctl, ['pause', 'dummy']) assert 'Failed' in result.output with patch('requests.patch', Mock(return_value=MockResponse(200))),\ patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): - result = self.runner.invoke(ctl, ['disable', 'dummy']) + result = self.runner.invoke(ctl, ['pause', 'dummy']) assert 'Cluster is already paused' in result.output @patch('patroni.ctl.get_dcs') From a9a70d44e2e138c6e4667ca575f289da59371f46 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 29 Aug 2016 15:04:50 +0200 Subject: [PATCH 18/30] Make the cached role coherrent with the actual one. When observing the leader running a master role, set the cached role stored in the state_handler to master as well. Failure to do so resulted in the manually promoted node to continue running with a cached 'replica' role. This led to the failure to create replication slots for the new replicas. We could do it conditionally, but both reading and writing the role require the same lock, and the unconditional approach makes the unit tests simpler. --- patroni/ha.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/patroni/ha.py b/patroni/ha.py index 75be52a9..c85afa42 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -158,6 +158,9 @@ class Ha(object): def enforce_master_role(self, message, promote_message): if self.state_handler.is_leader() or self.state_handler.role == 'master': + # Inform the state handler about its master role. + # It may be unaware of it if postgres is promoted manually. + self.state_handler.set_role('master') return message else: self.state_handler.promote() From 6dc1d9c88eae163ece5292fa4b8696f99bf9e754 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 29 Aug 2016 15:37:20 +0200 Subject: [PATCH 19/30] Trigger reinitialize from api and make it possible to reinitialize in a pause state --- patroni/api.py | 25 +++++---------------- patroni/ha.py | 55 ++++++++++++++++------------------------------- tests/test_api.py | 11 ++-------- tests/test_ha.py | 25 ++++++++------------- 4 files changed, 35 insertions(+), 81 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 7e066289..26c15cea 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -255,27 +255,12 @@ class RestApiHandler(BaseHTTPRequestHandler): @check_auth def do_POST_reinitialize(self): - patroni = self.server.patroni - cluster = patroni.dcs.get_cluster() - status_code = 500 - if cluster.is_paused(): - self._write_response(status_code, "Can't do reinitialize in the paused state") - return - - if cluster.is_unlocked(): - status_code = 503 - data = 'Cluster has no leader, can not reinitialize' - elif cluster.leader.name == patroni.ha.state_handler.name: - status_code = 503 - data = 'I am the leader, can not reinitialize' + data = self.server.patroni.ha.reinitialize() + if data is None: + status_code = 200 + data = 'reinitialize started' else: - action = patroni.ha.schedule_reinitialize() - if action is not None: - status_code = 503 - data = action + ' already in progress' - else: - status_code = 200 - data = 'reinitialize scheduled' + status_code = 503 self._write_response(status_code, data) def poll_failover_result(self, leader, candidate): diff --git a/patroni/ha.py b/patroni/ha.py index 75be52a9..7ef196f6 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -454,10 +454,6 @@ class Ha(object): logger.info("not proceeding with the restart: %s", reason_to_cancel) return False - def schedule(self, action, immediate=False): - with self._async_executor: - return self._async_executor.schedule(action, immediate) - def schedule_future_restart(self, restart_data): with self._async_executor: if not self.patroni.scheduled_restart: @@ -479,15 +475,6 @@ class Ha(object): return self.patroni.scheduled_restart.copy() if (self.patroni.scheduled_restart and isinstance(self.patroni.scheduled_restart, dict)) else None - def schedule_reinitialize(self): - return self.schedule('reinitialize') - - def reinitialize_scheduled(self): - return self._async_executor.scheduled_action == 'reinitialize' - - def schedule_restart(self, immediate=False): - return self.schedule('restart', immediate) - def restart_scheduled(self): return self._async_executor.scheduled_action == 'restart' @@ -500,7 +487,7 @@ class Ha(object): return (False, "restart conditions are not satisfied") with self._async_executor: - prev = self.schedule_restart(immediate=(not run_async)) + prev = self._async_executor.schedule('restart', not run_async) if prev is not None: return (False, prev + ' already in progress') if not run_async: @@ -512,28 +499,29 @@ class Ha(object): self._async_executor.run_async(self.state_handler.restart) return (True, "restart initiated") - def reinitialize(self, cluster): + def _do_reinitialize(self, cluster): self.state_handler.stop('immediate') self.state_handler.remove_data_directory() - clone_member = cluster.get_clone_member() - member_role = 'leader' if clone_member == cluster.leader else 'replica' + clone_member = self.cluster.get_clone_member() + member_role = 'leader' if clone_member == self.cluster.leader else 'replica' self.clone(clone_member, "from {0} '{1}'".format(member_role, clone_member.name)) - def process_scheduled_action(self): - if self.reinitialize_scheduled(): - if self.is_paused(): - logger.warning('Cluster is in a pause state, can not reinitialize') - self._async_executor.reset_scheduled_action() - elif self.cluster.is_unlocked(): - logger.error('Cluster has no leader, can not reinitialize') - self._async_executor.reset_scheduled_action() - elif self.has_lock(): - logger.error('I am the leader, can not reinitialize') - self._async_executor.reset_scheduled_action() - else: - self._async_executor.run_async(self.reinitialize, args=(self.cluster, )) - return 'reinitialize started' + def reinitialize(self): + with self._async_executor: + self.load_cluster_from_dcs() + + if self.cluster.is_unlocked(): + return 'Cluster has no leader, can not reinitialize' + + if self.cluster.leader.name == self.state_handler.name: + return 'I am the leader, can not reinitialize' + + action = self._async_executor.schedule('reinitialize', immediately=True) + if action is not None: + return '{0} already in progress'.format(action) + + self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, )) def handle_long_action_in_progress(self): if self.has_lock(): @@ -585,11 +573,6 @@ class Ha(object): if msg is not None: return msg - # currently it can trigger only reinitialize - msg = self.process_scheduled_action() - if msg is not None: - return msg - # is data directory empty? if self.state_handler.data_directory_empty(): return self.bootstrap() # new node diff --git a/tests/test_api.py b/tests/test_api.py index ad96ed87..1c025b15 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -40,7 +40,7 @@ class MockHa(object): state_handler = MockPostgresql() @staticmethod - def schedule_reinitialize(): + def reinitialize(): return 'reinitialize' @staticmethod @@ -238,15 +238,8 @@ class TestRestApiHandler(unittest.TestCase): cluster.is_paused.return_value = False request = 'POST /reinitialize HTTP/1.0' + self._authorization MockRestApiServer(RestApiHandler, request) - cluster.is_unlocked.return_value = False - MockRestApiServer(RestApiHandler, request) - with patch.object(MockHa, 'schedule_reinitialize', Mock(return_value=None)): + with patch.object(MockHa, 'reinitialize', Mock(return_value=None)): MockRestApiServer(RestApiHandler, request) - cluster.leader.name = 'test' - self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) - - cluster.is_paused.return_value = True - self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) @patch('time.sleep', Mock()) def test_RestApiServer_query(self): diff --git a/tests/test_ha.py b/tests/test_ha.py index 69293cec..52b7af18 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -277,31 +277,24 @@ class TestHa(unittest.TestCase): self.assertRaises(PostgresException, self.ha.bootstrap) def test_reinitialize(self): - self.ha.schedule_reinitialize() - self.ha.schedule_reinitialize() - self.ha.run_cycle() + self.assertIsNotNone(self.ha.reinitialize()) self.assertIsNone(self.ha._async_executor.scheduled_action) - with patch.object(Ha, 'is_paused', true): - self.ha.schedule_reinitialize() - self.ha.run_cycle() - self.assertIsNone(self.ha._async_executor.scheduled_action) - self.ha.cluster = get_cluster_initialized_with_leader() - self.ha.has_lock = true - self.ha.schedule_reinitialize() - self.ha.run_cycle() - self.assertIsNone(self.ha._async_executor.scheduled_action) + self.assertIsNone(self.ha.reinitialize()) + self.assertIsNotNone(self.ha._async_executor.scheduled_action) - self.ha.has_lock = false - self.ha.schedule_reinitialize() - self.ha.run_cycle() + self.assertIsNotNone(self.ha.reinitialize()) + + self.ha.state_handler.name = self.ha.cluster.leader.name + self.assertIsNotNone(self.ha.reinitialize()) def test_restart(self): self.assertEquals(self.ha.restart(), (True, 'restarted successfully')) self.p.restart = false self.assertEquals(self.ha.restart(), (False, 'restart failed')) - self.ha.schedule_reinitialize() + self.ha.cluster = get_cluster_initialized_with_leader() + self.ha.reinitialize() self.assertEquals(self.ha.restart(), (False, 'reinitialize already in progress')) with patch.object(self.ha, "restart_matches", return_value=False): self.assertEquals(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied")) From 366ed9cc522ac20541cad03f7a41e8df02b2a24e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 29 Aug 2016 15:39:24 +0200 Subject: [PATCH 20/30] fix pep8 formatting and implement missing tests --- features/patroni_api.feature | 4 ++-- patroni/ctl.py | 10 +++++----- patroni/dcs/__init__.py | 2 +- tests/test_patroni.py | 3 ++- 4 files changed, 10 insertions(+), 9 deletions(-) diff --git a/features/patroni_api.feature b/features/patroni_api.feature index b9199b00..3febb376 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -66,7 +66,7 @@ Scenario: check the failover via the API Given I run patronictl.py failover batman --master postgres0 --candidate postgres1 --force Then I receive a response returncode 0 And postgres1 is a leader after 5 seconds - And postgres1 role is the primary after 5 seconds + And postgres1 role is the primary after 10 seconds And postgres0 role is the secondary after 10 seconds And replication works from postgres1 to postgres0 after 20 seconds @@ -74,7 +74,7 @@ Scenario: check the scheduled failover Given I issue a scheduled failover from postgres1 to postgres0 in 1 seconds Then I receive a response returncode 0 And postgres0 is a leader after 20 seconds - And postgres0 role is the primary after 5 seconds + And postgres0 role is the primary after 10 seconds And postgres1 role is the secondary after 10 seconds And replication works from postgres0 to postgres1 after 25 seconds diff --git a/patroni/ctl.py b/patroni/ctl.py index 4288089a..a57c27fb 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -812,15 +812,15 @@ def flush(cluster_name, member_names, config_file, dcs, force, role, target): def toggle_pause(config_file, cluster_name, dcs, paused): config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) if cluster.is_paused() == paused: - raise PatroniCtlException("Cluster " + ("is already" if paused else "is not") + " paused") + raise PatroniCtlException('Cluster is {0} paused'.format(paused and 'already' or 'not')) - r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': paused}, auth_header(config)) + r = request_patroni(cluster.leader.member, 'patch', 'config', {'pause': paused or None}, auth_header(config)) if r.status_code == 200: - click.echo("Success: cluster management is " + ("paused" if paused else "resumed")) + click.echo('Success: cluster management is {0}'.format(paused and 'paused' or 'resumed')) else: - click.echo("Failed: " + ("pause" if paused else "resume") - + " cluster management status code={0}, ({1})".format(r.status_code, r.text)) + click.echo('Failed: {0} cluster management status code={1}, ({2})'.format( + paused and 'pause' or 'resume', r.status_code, r.text)) @ctl.command('pause', help='Disable auto failover') diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index bbeab3c5..0c403a1c 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -229,7 +229,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader def is_paused(self): - return self.config and self.config.data.get('pause', False) + return self.config and self.config.data.get('pause', False) or False @six.add_metaclass(abc.ABCMeta) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index f9f135cc..d3963a1f 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -54,7 +54,8 @@ class TestPatroni(unittest.TestCase): with patch.object(Patroni, 'run', Mock(side_effect=SleepException)): self.assertRaises(SleepException, _main) with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())): - _main() + with patch('patroni.ha.Ha.is_paused', Mock(return_value=True)): + _main() @patch('patroni.config.Config.save_cache', Mock()) @patch('patroni.config.Config.reload_local_configuration', Mock(return_value=True)) From 1374fb3a2d1e649205ac0125c8ae5ffdd07a3103 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 30 Aug 2016 10:30:28 +0200 Subject: [PATCH 21/30] Set role to uninitialized when removing data directory --- patroni/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index da4b932a..84960b67 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -799,7 +799,6 @@ class Postgresql(object): else: logger.error('unable to rewind the former master') self.remove_data_directory() - self.set_role('uninitialized') ret = True self._need_rewind = False else: @@ -950,6 +949,7 @@ $$""".format(name, ' '.join(options)), name, password, password) logger.exception("Could not rename data directory %s", self._data_dir) def remove_data_directory(self): + self.set_role('uninitialized') logger.info('Removing data directory: %s', self._data_dir) try: if os.path.islink(self._data_dir): From 0afdb816ba3652029908d90caac3987f8fa58184 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 30 Aug 2016 10:38:40 +0200 Subject: [PATCH 22/30] Unfinished promote may not break paused cluster. When a node to promote dies before finishing the promote and the cluster is in a standby mode, the failover key sticks indefinitely, preventing any master to take over the leader role. Prevent it by letting the node in a master role cleanup the failover key if the node to failover is not present among the members. The master check cannot be performed by the node role alone, since the node will not change its cached role on a manual promote. We need to check the DB state as well. --- patroni/ha.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/patroni/ha.py b/patroni/ha.py index 60734ee6..5d106512 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -237,6 +237,11 @@ class Ha(object): if failover.candidate == self.state_handler.name: # manual failover to me return True elif self.is_paused(): + # Remove failover key if the node to failover has terminated to avoid waiting for it indefinitely + # In order to avoid race conditions only the master is allowed to do so. + if (not self.cluster.get_member(failover.candidate, fallback_to_leader=False) and + (self.state_handler.is_leader() or self.state_handler.role == 'master')): + self.dcs.manual_failover('', '', index=self.cluster.failover.index) return False # find specific node and check that it is healthy From 11359a26a919802236f7d75052a06c2a76219760 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 30 Aug 2016 12:00:51 +0200 Subject: [PATCH 23/30] Improve incomplete failover is a paused mode. Instead of empying the stale failover key as a master and bailing out, continue with the healthiest node evaluation. This should make the actual master acquire the leader key faster. Emit the warning message as well and add unit tests. --- patroni/ha.py | 11 ++++++++--- tests/test_ha.py | 3 +++ 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 5d106512..26565f2f 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -238,10 +238,12 @@ class Ha(object): return True elif self.is_paused(): # Remove failover key if the node to failover has terminated to avoid waiting for it indefinitely - # In order to avoid race conditions only the master is allowed to do so. + # In order to avoid attempts to delete this key from all nodes only the master is allowed to do it. if (not self.cluster.get_member(failover.candidate, fallback_to_leader=False) and - (self.state_handler.is_leader() or self.state_handler.role == 'master')): + self.state_handler.is_leader()): + logger.warning("manual failover: removing failover key because failover candidate is not running") self.dcs.manual_failover('', '', index=self.cluster.failover.index) + return None return False # find specific node and check that it is healthy @@ -281,7 +283,9 @@ class Ha(object): def is_healthiest_node(self): if self.is_paused() and not self.patroni.nofailover and \ self.cluster.failover and not self.cluster.failover.scheduled_at: - return self.manual_failover_process_no_leader() + ret = self.manual_failover_process_no_leader() + if ret is not None: # continue if we just deleted the stale failover key as a master + return ret if self.state_handler.is_leader(): # leader is always the healthiest return True @@ -377,6 +381,7 @@ class Ha(object): def process_unhealthy_cluster(self): """Cluster has no leader key""" + if self.is_healthiest_node(): if self.acquire_lock(): if self.cluster.failover: diff --git a/tests/test_ha.py b/tests/test_ha.py index 52b7af18..dd717771 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -394,6 +394,9 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock') self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', '', None)) self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock') + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', 'blabla', None)) + self.p.is_leader = true + self.assertEquals('PAUSE: acquired session lock as a leader', self.ha.run_cycle()) def test_is_healthiest_node(self): self.ha.state_handler.is_leader = false From 8028877be0b1f92b4825d50aa449abc8eb9a7a8e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 30 Aug 2016 16:49:28 +0200 Subject: [PATCH 24/30] Remove failover key only after becoming master --- patroni/ha.py | 25 ++++++++++++++++--------- tests/test_ha.py | 4 ++++ 2 files changed, 20 insertions(+), 9 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 60734ee6..80645320 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -147,6 +147,7 @@ class Ha(object): node_to_follow = self._get_node_to_follow(self.cluster) if self.is_paused(): + self.state_handler.set_role('master' if is_leader else 'replica') if is_leader: return 'continue to run as master without lock' elif not node_to_follow: @@ -328,7 +329,6 @@ class Ha(object): logger.warning('Found a stale %s value, cleaning up: %s', action_name, scheduled_at.isoformat()) cleanup_fn() - self.dcs.manual_failover('', '', index=self.cluster.failover.index) return False # The value is very close to now @@ -367,16 +367,21 @@ class Ha(object): logger.warning('manual failover: leader name does not match: %s != %s', failover.leader, self.state_handler.name) - logger.info('Trying to clean up failover key') + logger.info('Cleaning up failover key') self.dcs.manual_failover('', '', index=failover.index) def process_unhealthy_cluster(self): """Cluster has no leader key""" if self.is_healthiest_node(): if self.acquire_lock(): - if self.cluster.failover: - logger.info('Cleaning up failover key after acquiring leader lock...') - self.dcs.manual_failover('', '') + failover = self.cluster.failover + if failover: + if self.is_paused() and failover.leader and failover.candidate: + logger.info('Updating failover key after acquiring leader lock...') + self.dcs.manual_failover('', failover.candidate, failover.scheduled_at, failover.index) + else: + logger.info('Cleaning up failover key after acquiring leader lock...') + self.dcs.manual_failover('', '') self.load_cluster_from_dcs() return self.enforce_master_role('acquired session lock as a leader', 'promoted self to leader by acquiring session lock') @@ -392,7 +397,7 @@ class Ha(object): def process_healthy_cluster(self): if self.has_lock(): - if self.cluster.failover: + if self.cluster.failover and (not self.is_paused() or self.state_handler.is_leader()): msg = self.process_manual_failover_from_leader() if msg is not None: return msg @@ -589,7 +594,6 @@ class Ha(object): self.state_handler.name, self.cluster.initialize, self.state_handler.sysid) sys.exit(1) - # try to start dead postgres if not self.state_handler.is_healthy(): if self.is_paused(): if self.has_lock(): @@ -598,6 +602,8 @@ class Ha(object): return 'removed leader lock because postgres is not running' else: return 'postgres is not running' + + # try to start dead postgres msg = self.recover() if msg is not None: return msg @@ -625,8 +631,9 @@ class Ha(object): return 'demoted self because DCS is not accessible and i was a leader' return 'DCS is not accessible' except (psycopg2.Error, PostgresConnectionException): - logger.exception('Error communicating with PostgreSQL. Will try again later') + return 'Error communicating with PostgreSQL. Will try again later' def run_cycle(self): with self._async_executor: - return (self.is_paused() and 'PAUSE: ' or '') + self._run_cycle() + info = self._run_cycle() + return (self.is_paused() and 'PAUSE: ' or '') + info diff --git a/tests/test_ha.py b/tests/test_ha.py index 52b7af18..ec2b8180 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -394,6 +394,10 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock') self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', '', None)) self.assertEquals(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock') + self.p.is_leader = false + self.p.set_role('replica') + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', self.p.name, None)) + self.assertEquals(self.ha.run_cycle(), 'PAUSE: promoted self to leader by acquiring session lock') def test_is_healthiest_node(self): self.ha.state_handler.is_leader = false From 1dcdd6eaa0c659043c4e7dfe09a7f2a7f2d11476 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 30 Aug 2016 16:50:07 +0200 Subject: [PATCH 25/30] Acceptance tests for pause mode --- features/patroni_api.feature | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/features/patroni_api.feature b/features/patroni_api.feature index 3febb376..e201613c 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -45,9 +45,11 @@ Scenario: check dynamic configuration change via DCS Then I receive a response code 200 And I receive a response tags {'tag': 'new_value'} -Scenario: check API requests for the primary-replica pair - Given I start postgres1 - And replication works from postgres0 to postgres1 after 20 seconds +Scenario: check API requests for the primary-replica pair in the pause mode + Given I run patronictl.py pause batman + Then I receive a response returncode 0 + When I start postgres1 + Then replication works from postgres0 to postgres1 after 20 seconds When I issue a GET request to http://127.0.0.1:8009/replica Then I receive a response code 200 And I receive a response state running @@ -62,7 +64,7 @@ Scenario: check API requests for the primary-replica pair When I sleep for 10 seconds Then postgres1 role is the secondary after 15 seconds -Scenario: check the failover via the API +Scenario: check the failover via the API in the pause mode Given I run patronictl.py failover batman --master postgres0 --candidate postgres1 --force Then I receive a response returncode 0 And postgres1 is a leader after 5 seconds @@ -71,6 +73,11 @@ Scenario: check the failover via the API And replication works from postgres1 to postgres0 after 20 seconds Scenario: check the scheduled failover + Given I issue a scheduled failover from postgres1 to postgres0 in 1 seconds + Then I receive a response returncode 1 + And I receive a response output "Can't schedule failover in the paused state" + When I run patronictl.py resume batman + Then I receive a response returncode 0 Given I issue a scheduled failover from postgres1 to postgres0 in 1 seconds Then I receive a response returncode 0 And postgres0 is a leader after 20 seconds @@ -84,7 +91,7 @@ Scenario: check the scheduled restart And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 5 seconds Given I issue a scheduled restart at http://127.0.0.1:8008 in 1 seconds with {"role": "replica"} Then I receive a response code 202 - And I sleep for 10 seconds + And I sleep for 2 seconds And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 10 seconds Given I issue a scheduled restart at http://127.0.0.1:8008 in 1 seconds with {"restart_pending": "True"} Then I receive a response code 202 From 4d72eef164232e5d52d4301aa0b105263ac52eb0 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 31 Aug 2016 12:38:02 +0200 Subject: [PATCH 26/30] Execute API restart outside of lock Otherwise it was blocking HA loop... --- patroni/async_executor.py | 7 +------ patroni/ha.py | 20 ++++++++++---------- tests/test_ha.py | 28 ++++++++++++++-------------- 3 files changed, 25 insertions(+), 30 deletions(-) diff --git a/patroni/async_executor.py b/patroni/async_executor.py index 640ef992..04b6a5a8 100644 --- a/patroni/async_executor.py +++ b/patroni/async_executor.py @@ -7,21 +7,19 @@ logger = logging.getLogger(__name__) class AsyncExecutor(object): def __init__(self): - self._busy = False self._thread_lock = RLock() self._scheduled_action = None self._scheduled_action_lock = RLock() @property def busy(self): - return self._busy + return self.scheduled_action is not None def schedule(self, action, immediately=False): with self._scheduled_action_lock: if self._scheduled_action is not None: return self._scheduled_action self._scheduled_action = action - self._busy = immediately return None @property @@ -32,7 +30,6 @@ class AsyncExecutor(object): def reset_scheduled_action(self): with self._scheduled_action_lock: self._scheduled_action = None - self._busy = False def run(self, func, args=()): try: @@ -41,11 +38,9 @@ class AsyncExecutor(object): logger.exception('Exception during execution of long running task %s', self.scheduled_action) finally: with self: - self._busy = False self.reset_scheduled_action() def run_async(self, func, args=()): - self._busy = True Thread(target=self.run, args=(func, args)).start() def __enter__(self): diff --git a/patroni/ha.py b/patroni/ha.py index 11f02308..cc9bca0f 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -505,17 +505,17 @@ class Ha(object): return (False, "restart conditions are not satisfied") with self._async_executor: - prev = self._async_executor.schedule('restart', not run_async) + prev = self._async_executor.schedule('restart') if prev is not None: return (False, prev + ' already in progress') - if not run_async: - if self._async_executor.run(self.state_handler.restart): - return (True, 'restarted successfully') - else: - return (False, 'restart failed') - else: - self._async_executor.run_async(self.state_handler.restart) - return (True, "restart initiated") + + if run_async: + self._async_executor.run_async(self.state_handler.restart) + return (True, 'restart initiated') + elif self._async_executor.run(self.state_handler.restart): + return (True, 'restarted successfully') + else: + return (False, 'restart failed') def _do_reinitialize(self, cluster): self.state_handler.stop('immediate') @@ -539,7 +539,7 @@ class Ha(object): if action is not None: return '{0} already in progress'.format(action) - self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, )) + self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, )) def handle_long_action_in_progress(self): if self.has_lock(): diff --git a/tests/test_ha.py b/tests/test_ha.py index cf4a84e5..754a5788 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -4,7 +4,7 @@ import os import pytz import unittest -from mock import Mock, MagicMock, patch +from mock import Mock, MagicMock, PropertyMock, patch from patroni.config import Config from patroni.dcs import Cluster, Failover, Leader, Member, get_dcs from patroni.dcs.etcd import Client @@ -90,7 +90,7 @@ zookeeper: 'postmaster_start_time': str(postmaster_start_time)} -def run_async(func, args=()): +def run_async(self, func, args=()): return func(*args) if args else func() @@ -109,6 +109,8 @@ def run_async(func, args=()): @patch.object(etcd.Client, 'write', etcd_write) @patch.object(etcd.Client, 'read', etcd_read) @patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException)) +@patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False)) +@patch('patroni.async_executor.AsyncExecutor.run_async', run_async) @patch('subprocess.call', Mock(return_value=0)) class TestHa(unittest.TestCase): @@ -131,7 +133,6 @@ class TestHa(unittest.TestCase): self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test', 'name': 'foo', 'retry_timeout': 10}}) self.ha = Ha(MockPatroni(self.p, self.e)) - self.ha._async_executor.run_async = run_async self.ha.old_cluster = self.e.get_cluster() self.ha.cluster = get_cluster_not_initialized_without_leader() self.ha.load_cluster_from_dcs = Mock() @@ -278,11 +279,9 @@ class TestHa(unittest.TestCase): def test_reinitialize(self): self.assertIsNotNone(self.ha.reinitialize()) - self.assertIsNone(self.ha._async_executor.scheduled_action) self.ha.cluster = get_cluster_initialized_with_leader() self.assertIsNone(self.ha.reinitialize()) - self.assertIsNotNone(self.ha._async_executor.scheduled_action) self.assertIsNotNone(self.ha.reinitialize()) @@ -300,18 +299,19 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied")) def test_restart_in_progress(self): - self.ha._async_executor.schedule('restart', True) - self.assertTrue(self.ha.restart_scheduled()) - self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race') + with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)): + self.ha.restart(run_async=True) + self.assertTrue(self.ha.restart_scheduled()) + self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race') - self.ha.cluster = get_cluster_initialized_with_leader() - self.assertEquals(self.ha.run_cycle(), 'restart in progress') + self.ha.cluster = get_cluster_initialized_with_leader() + self.assertEquals(self.ha.run_cycle(), 'restart in progress') - self.ha.has_lock = true - self.assertEquals(self.ha.run_cycle(), 'updated leader lock during restart') + self.ha.has_lock = true + self.assertEquals(self.ha.run_cycle(), 'updated leader lock during restart') - self.ha.update_lock = false - self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart') + self.ha.update_lock = false + self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart') @patch('requests.get', requests_get) @patch('time.sleep', Mock()) From 0e8220f9f2271257405552f68a97b48db0ffc9b8 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 31 Aug 2016 15:08:31 +0200 Subject: [PATCH 27/30] BUGFIX: dcs configuration need to be updated from dcs... loop_wait and ttl is configured cluster-wide via config key stored in DCS. Depending on values of these parameters we are configuring different kind of timeouts used in DCS controllers. In order to fetch this configuration we first need to create DCS controller and only after apply parameters... --- patroni/__init__.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 6eba3790..d713969b 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -38,9 +38,11 @@ class Patroni(object): try: cluster = self.dcs.get_cluster() if cluster and cluster.config: - self.config.set_dynamic_configuration(cluster.config) + if self.config.set_dynamic_configuration(cluster.config): + self.dcs.reload_config(self.config) elif not self.config.dynamic_configuration and 'bootstrap' in self.config: - self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']) + if self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']): + self.dcs.reload_config(self.config) break except DCSError: logger.warning('Can not get cluster from dcs') From 33ff372ef6b1e56410f008a463329da1b32482be Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 1 Sep 2016 11:08:26 +0200 Subject: [PATCH 28/30] Always try to rewind on manual failover --- features/patroni_api.feature | 6 +++--- patroni/__init__.py | 4 ++-- patroni/dcs/__init__.py | 3 +++ patroni/ha.py | 32 +++++++++++++++++++++----------- patroni/postgresql.py | 35 +++++++++++++++++------------------ tests/test_ha.py | 4 ++++ tests/test_postgresql.py | 8 ++------ 7 files changed, 52 insertions(+), 40 deletions(-) diff --git a/features/patroni_api.feature b/features/patroni_api.feature index e201613c..67b886e7 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -34,13 +34,13 @@ Scenario: check local configuration reload Then I receive a response code 202 Scenario: check dynamic configuration change via DCS - Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "loop_wait": 1, "postgresql": {"parameters": {"max_connections": 101}}} + Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "loop_wait": 2, "postgresql": {"parameters": {"max_connections": 101}}} Then I receive a response code 200 - And I receive a response loop_wait 1 + And I receive a response loop_wait 2 And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds When I issue a GET request to http://127.0.0.1:8008/config Then I receive a response code 200 - And I receive a response loop_wait 1 + And I receive a response loop_wait 2 When I issue a GET request to http://127.0.0.1:8008/patroni Then I receive a response code 200 And I receive a response tags {'tag': 'new_value'} diff --git a/patroni/__init__.py b/patroni/__init__.py index d713969b..7d7a38c2 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -53,7 +53,7 @@ class Patroni(object): @property def nofailover(self): - return self.tags.get('nofailover', False) + return bool(self.tags.get('nofailover', False)) def reload_config(self): try: @@ -78,7 +78,7 @@ class Patroni(object): @property def noloadbalance(self): - return self.tags.get('noloadbalance', False) + return bool(self.tags.get('noloadbalance', False)) def schedule_next_run(self): self.next_run += self.dcs.loop_wait diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 0c403a1c..6b8aaaec 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -186,6 +186,9 @@ class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')): return Failover(index, data.get('leader'), data.get('member'), data.get('scheduled_at')) + def __len__(self): + return int(bool(self.leader)) + int(bool(self.candidate)) + class ClusterConfig(namedtuple('ClusterConfig', 'index,data,modify_index')): diff --git a/patroni/ha.py b/patroni/ha.py index cc9bca0f..d974951b 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -134,7 +134,7 @@ class Ha(object): return node_to_follow if node_to_follow and node_to_follow.name != self.state_handler.name else None - def follow(self, demote_reason, follow_reason, refresh=True, recovery=False): + def follow(self, demote_reason, follow_reason, refresh=True, recovery=False, need_rewind=None): if refresh: self.load_cluster_from_dcs() @@ -146,14 +146,14 @@ class Ha(object): node_to_follow = self._get_node_to_follow(self.cluster) - if self.is_paused(): + if self.is_paused() and not self.state_handler.need_rewind: self.state_handler.set_role('master' if is_leader else 'replica') if is_leader: return 'continue to run as master without lock' elif not node_to_follow: return 'no action' - self.state_handler.follow(node_to_follow, self.cluster.leader, recovery, self._async_executor) + self.state_handler.follow(node_to_follow, self.cluster.leader, recovery, self._async_executor, need_rewind) return ret @@ -307,14 +307,14 @@ class Ha(object): def demote(self, delete_leader=True): if delete_leader: self.state_handler.stop() - self.state_handler.set_role('unknown') + self.state_handler.set_role('demoted') self.dcs.delete_leader() self.touch_member() self.dcs.reset_cluster() - sleep(2) # Give a time to somebody to promote + sleep(2) # Give a time to somebody to take the leader lock cluster = self.dcs.get_cluster() node_to_follow = self._get_node_to_follow(cluster) - self.state_handler.follow(node_to_follow, cluster.leader, True) + self.state_handler.follow(node_to_follow, cluster.leader, recovery=True, need_rewind=True) else: self.state_handler.follow(None, None) @@ -399,11 +399,18 @@ class Ha(object): return self.follow('demoted self after trying and failing to obtain lock', 'following new leader after trying and failing to obtain lock') else: + # when we are doing manual failover there is no guaranty that new leader is ahead of any other node + need_rewind = bool(self.cluster.failover) or self.patroni.nofailover + if need_rewind: + sleep(2) # Give a time to somebody to take the leader lock + if self.patroni.nofailover: return self.follow('demoting self because I am not allowed to become master', - 'following a different leader because I am not allowed to promote') - return self.follow('demoting self because i am not the healthiest node', # should not happen in real life - 'following a different leader because i am not the healthiest node') + 'following a different leader because I am not allowed to promote', + need_rewind=need_rewind) + return self.follow('demoting self because i am not the healthiest node', + 'following a different leader because i am not the healthiest node', + need_rewind=need_rewind) def process_healthy_cluster(self): if self.has_lock(): @@ -413,6 +420,9 @@ class Ha(object): return msg if self.is_paused() and not self.state_handler.is_leader(): + if self.cluster.failover and self.cluster.failover.candidate == self.state_handler.name: + return 'waiting to become master after promote...' + self.dcs.delete_leader() self.dcs.reset_cluster() return 'removed leader lock because postgres is not running as master' @@ -585,7 +595,7 @@ class Ha(object): return self.handle_long_action_in_progress() # we've got here, so any async action has finished. Check if we tried to recover and failed - if self.recovering: + if self.recovering and not self.state_handler.need_rewind: self.recovering = False msg = self.post_recover() if msg is not None: @@ -610,7 +620,7 @@ class Ha(object): self.dcs.delete_leader() self.dcs.reset_cluster() return 'removed leader lock because postgres is not running' - else: + elif not self.state_handler.need_rewind: return 'postgres is not running' # try to start dead postgres diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 84960b67..99d8cde0 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -729,7 +729,14 @@ class Postgresql(object): except OSError: logger.exception("Unable to list %s", status_dir) - def follow(self, member, leader, recovery=False, async_executor=None): + @property + def need_rewind(self): + return self._need_rewind + + def follow(self, member, leader, recovery=False, async_executor=None, need_rewind=None): + if need_rewind is not None: + self._need_rewind = need_rewind + primary_conninfo = self.primary_conninfo(member) if self.check_recovery_conf(primary_conninfo) and not recovery: @@ -742,31 +749,23 @@ class Postgresql(object): self._do_follow(primary_conninfo, leader, recovery) def _do_follow(self, primary_conninfo, leader, recovery=False): - change_role = self.role == 'master' + change_role = self.role in ('master', 'demoted') - if change_role: - if leader: - if leader.name == self.name: - self._need_rewind = False - primary_conninfo = None - if self.is_running(): - return - else: - self._need_rewind = bool(leader.conn_url) and self.can_rewind - else: - self._need_rewind = False - primary_conninfo = None + if leader and leader.name == self.name: + primary_conninfo = None + self._need_rewind = False + if self.is_running(): + return + + self._need_rewind &= bool(leader and leader.conn_url) and self.can_rewind if self._need_rewind: - logger.info("set the rewind flag after demote") + logger.info("rewind flag is set") self.set_role('unknown') if self.is_running() and not self.stop(): return logger.warning('Can not run pg_rewind because postgres is still running') - if not (leader and leader.conn_url): - return logger.info('Leader unknown, can not rewind') - # prepare pg_rewind connection r = leader.conn_kwargs(self._superuser) diff --git a/tests/test_ha.py b/tests/test_ha.py index 754a5788..3cf29d9f 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -364,6 +364,7 @@ class TestHa(unittest.TestCase): self.assertEquals('PAUSE: no action. i am the leader with the lock', self.ha.run_cycle()) @patch('requests.get', requests_get) + @patch('time.sleep', Mock()) def test_manual_failover_process_no_leader(self): self.p.is_leader = false self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', self.p.name, None)) @@ -388,6 +389,7 @@ class TestHa(unittest.TestCase): self.ha.patroni.nofailover = True self.assertEquals(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote') + @patch('time.sleep', Mock()) def test_manual_failover_process_no_leader_in_pause(self): self.ha.is_paused = true self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None)) @@ -491,6 +493,8 @@ class TestHa(unittest.TestCase): self.p.name = 'leader' self.ha.cluster = get_cluster_initialized_with_leader() self.assertEquals(self.ha.run_cycle(), 'PAUSE: removed leader lock because postgres is not running as master') + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', self.p.name, None)) + self.assertEquals(self.ha.run_cycle(), 'PAUSE: waiting to become master after promote...') def test_postgres_unhealthy_in_pause(self): self.ha.is_paused = true diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 2ab29124..9fb7bb7a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -263,20 +263,16 @@ class TestPostgresql(unittest.TestCase): with patch.object(Postgresql, 'restart', Mock(return_value=False)): self.p.set_role('replica') self.p.follow(None, None) # restart without rewind - self.p.set_role('master') with patch.object(Postgresql, 'stop', Mock(return_value=False)): - self.p.follow(self.leader, self.leader) # failed to stop postgres - - self.p.follow(self.leader, None) # Leader unknown, can not rewind + self.p.follow(self.leader, self.leader, need_rewind=True) # failed to stop postgres self.p.follow(self.leader, self.leader) # "leader" is not accessible or is_in_recovery with patch.object(Postgresql, 'checkpoint', Mock(return_value=None)): self.p.follow(self.leader, self.leader) - self.p.set_role('master') mock_pg_rewind.return_value = True - self.p.follow(self.leader, self.leader) + self.p.follow(self.leader, self.leader, need_rewind=True) self.p.follow(None, None) # check_recovery_conf... From f082ecf60b1cd62249782cc94d532450de87b7d1 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 1 Sep 2016 11:29:45 +0200 Subject: [PATCH 29/30] Set _need_rewind to True if the node was previously known as a master --- patroni/ha.py | 1 + patroni/postgresql.py | 2 ++ 2 files changed, 3 insertions(+) diff --git a/patroni/ha.py b/patroni/ha.py index d974951b..7a31c1fe 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -400,6 +400,7 @@ class Ha(object): 'following new leader after trying and failing to obtain lock') else: # when we are doing manual failover there is no guaranty that new leader is ahead of any other node + # node tagged as nofailover can be ahead of the new leader either, but it is always excluded from elections need_rewind = bool(self.cluster.failover) or self.patroni.nofailover if need_rewind: sleep(2) # Give a time to somebody to take the leader lock diff --git a/patroni/postgresql.py b/patroni/postgresql.py index ce8b6183..b023faf2 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -774,6 +774,8 @@ class Postgresql(object): self._need_rewind = False if self.is_running(): return + elif change_role: + self._need_rewind = True self._need_rewind &= bool(leader and leader.conn_url) and self.can_rewind From fef4e046e1ba2c2d9b9a825b9bb59bed3328a414 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 2 Sep 2016 09:00:03 +0200 Subject: [PATCH 30/30] Avoid setting the role to unknown during rewind. Previously, that was necessary in order to avoid repeating the rewind after failure. Nowadays, depending on the failure, we either want to retry (if PostgreSQL was not stopped on time or leader did not manage to acquire a master role yet), or won't retry at all if the leader is not available, assuming the replica role. In both cases, the hack with setting the role to unknown seems to be unnecessary and actually stops callbacks from running if rewind is done not from the first attempt. --- patroni/postgresql.py | 1 - 1 file changed, 1 deletion(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index b023faf2..70df0072 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -782,7 +782,6 @@ class Postgresql(object): if self._need_rewind: logger.info("rewind flag is set") - self.set_role('unknown') if self.is_running() and not self.stop(): return logger.warning('Can not run pg_rewind because postgres is still running')