diff --git a/features/patroni_api.feature b/features/patroni_api.feature index 6bbdbeb1..934e8b76 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -72,3 +72,16 @@ Scenario: check the scheduled failover And postgres0 role is the primary after 5 seconds And postgres1 role is the secondary after 10 seconds And replication works from postgres0 to postgres1 after 25 seconds + +Scenario: check the scheduled restart + Given I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"checkpoint_warning": "60s"}}} + Then I receive a response code 200 + 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 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 {"with_pending_restart_flag": "True"} + Then I receive a response code 202 + And Response on GET http://127.0.0.1:8008/patroni does not contain pending_restart after 10 seconds + diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index 90c38564..4d2f1709 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -103,18 +103,35 @@ def scheduled_failover(context, at_url, from_host, to_host, in_seconds): """.format(at_url, from_host, to_host, datetime.now(pytz.utc) + timedelta(seconds=int(in_seconds)))) +@step('I issue a scheduled restart at {url:url} in {in_seconds:d} seconds with {data}') +def scheduled_restart(context, url, in_seconds, data): + data = data and json.loads(data) or {} + restart_options = ['"schedule": "{0}"'.format((datetime.now(pytz.utc) + timedelta(seconds=int(in_seconds))).isoformat())] + for key in data: + if key == 'schedule': + continue + restart_options.append('"{0}": "{1}"'.format(key, data[key])) + context.execute_steps(u"""Given I issue a POST request to {0}/restart with {{{1}}}""". + format(url, ','.join(restart_options))) + + @step('I add tag {tag:w} {value:w} to {pg_name:w} config') def add_tag_to_config(context, tag, value, pg_name): context.pctl.add_tag_to_config(pg_name, tag, value) @then('Response on GET {url} contains {value} after {timeout:d} seconds') -def check_http_response(context, url, value, timeout): +def check_http_response(context, url, value, timeout, negate=False): for _ in range(int(timeout)): r = requests.get(url) - if value in r.content.decode('utf-8'): + if (value in r.content.decode('utf-8')) != negate: break time.sleep(1) else: assert False,\ - "Value {0} is not present in response after {1} seconds".format(value, timeout) + "Value {0} is {0} present in response after {1} seconds".format("not" if not negate else "", value, timeout) + + +@then('Response on GET {url} does not contain {value} after {timeout:d} seconds') +def check_not_in_http_response(context, url, value, timeout): + check_http_response(context, url, value, timeout, negate=True) diff --git a/patroni/api.py b/patroni/api.py index a6a14dbe..b91eea30 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -65,7 +65,8 @@ class RestApiHandler(BaseHTTPRequestHandler): response['pending_restart'] = True response['patroni'] = {'version': patroni.version, 'scope': patroni.postgresql.scope} if patroni.scheduled_restart and isinstance(patroni.scheduled_restart, dict): - response['scheduled_restart'] = patroni.scheduled_restart + response['scheduled_restart'] = patroni.scheduled_restart.copy() + response['scheduled_restart']['schedule'] = (response['scheduled_restart']['schedule']).isoformat() self._write_json_response(status_code, response) def do_GET(self, write_status_code_only=False): @@ -236,7 +237,7 @@ class RestApiHandler(BaseHTTPRequestHandler): @check_auth def do_DELETE_restart(self): - self.server.patroni.ha.delete_future_restart() + self.server.patroni.ha.delete_future_restart(take_lock=True) data = "scheduled restart deleted" code = 200 self._write_response(code, data) diff --git a/patroni/ha.py b/patroni/ha.py index 0b6f3b15..8c282c95 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -385,6 +385,36 @@ class Ha(object): return self.follow('demoting self because i do not have the lock and i was a leader', 'no action. i am a secondary and i am following a leader', False) + def evaluate_scheduled_restart(self): + # restart if we need to + restart_data = self.future_restart_scheduled() + if (restart_data and + self.should_run_scheduled_action('restart', restart_data['schedule'], self.delete_future_restart)): + try: + reason_to_cancel = "" + # checking the restart filters here seem to be less ugly than moving them into the + # run_scheduled_action. + if restart_data.get('role') and restart_data['role'] != self.state_handler.role: + reason_to_cancel = "host role mismatch" + + if (restart_data.get('postgres_version') and + self.state_hander.postgres_version_to_int(restart_data['postgres_version']) <= + int(self.state_hander.server_version)): + reason_to_cancel = "postgres version mismatch" + + if 'with_pending_restart_flag' in restart_data and not self.state_handler.pending_restart: + reason_to_cancel = "pending restart flag is not set" + + if not reason_to_cancel: + if self.state_handler.restart(): + logger.info("Scheduled restart successfull") + else: + logger.warning("Scheduled restart failed") + else: + logger.info("not proceeding with the scheduled restart: {0}".format(reason_to_cancel)) + finally: + self.delete_future_restart() + def schedule(self, action): with self._async_executor: return self._async_executor.schedule(action) @@ -397,13 +427,20 @@ class Ha(object): return True return False - def delete_future_restart(self): - with self._async_executor: + def delete_future_restart(self, take_lock=False): + if take_lock: + with self._async_executor: + self.patroni.scheduled_restart = {} + else: self.patroni.scheduled_restart = {} def immediate_restart_scheduled(self): return self._async_executor.scheduled_action == 'restart' + def future_restart_scheduled(self): + 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') @@ -517,6 +554,7 @@ class Ha(object): if self.cluster.is_unlocked(): return self.process_unhealthy_cluster() else: + self.evaluate_scheduled_restart() return self.process_healthy_cluster() finally: # we might not have a valid PostgreSQL connection here if another thread diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 4016c6d7..8951e4c3 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -912,3 +912,29 @@ $$""".format(name, ' '.join(options)), name, password, password) time.sleep(5) return ret + + @staticmethod + def postgres_version_to_int(pg_version): + """ Convert the server_version to integer + + >>> Postgresql.postgres_version_to_int('9.5.3') + 90503 + >>> Postgresql.postgres_version_to_int('9.3.13') + 90313 + >>> Postgresql.postgres_version_to_int('10.1') + 100100 + """ + components = pg_version.split('.') + + result = [] + if len(components) < 2 or len(components) > 3: + raise Exception("Invalid PostgreSQL format: X.Y or X.Y.Z is accepted: {0}".format(pg_version)) + if len(components) == 2: + # new style verion numbers, i.e. 10.1 + components.append('0') + try: + result = [c if int(c) > 10 else '0{0}'.format(c) for c in components] + result = int(''.join(result)) + except ValueError: + raise Exception("Exception when parsing PostgreSQL version: {0}".format(pg_version)) + return result diff --git a/tests/test_api.py b/tests/test_api.py index d0974e22..6b9f93de 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -3,6 +3,7 @@ import json import psycopg2 import unittest +import dateutil.parser from mock import Mock, patch from patroni.api import RestApiHandler, RestApiServer from patroni.dcs import ClusterConfig, Member @@ -66,7 +67,7 @@ class MockPatroni(object): tags = {} version = '0.00' noloadbalance = Mock(return_value=False) - scheduled_restart = {'schedule': '2016-08-29 12:45TZ+1'} + scheduled_restart = {'schedule': dateutil.parser.parse('2016-08-29 12:45TZ+1')} @staticmethod def sighup_handler():