From 854ff27e56873ec6c014e50c53c1920680334a84 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jun 2016 09:50:09 +0200 Subject: [PATCH] Allow normal conditional restarts. In addition, use the RLock instead of the Lock in async executor to make sure the lock can be acquired more than once from a single thread. --- patroni/api.py | 74 +++++++++++++++++++-------------------- patroni/async_executor.py | 7 ++-- patroni/ha.py | 28 ++++++++------- 3 files changed, 56 insertions(+), 53 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 3fbab534..c6b5e225 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -195,51 +195,51 @@ class RestApiHandler(BaseHTTPRequestHandler): status_code = 500 data = 'restart failed' request = self._read_json_content(body_is_optional=True) - if not request: # unconditional restart - try: - status, data = self.server.patroni.ha.restart() - status_code = 200 if status else 503 - except Exception: - logger.exception('Exception during restart') - else: - logger.info("received scheduled restart request: {0}".format(request)) - for k in request: - if k == 'schedule': - (_, data, request[k]) = self.parse_schedule(request[k], "restart") - if _: - status_code = _ - break - elif k == 'role': - if request[k] not in ('master', 'replica'): - status_code = 400 - data = "PostgreSQL role should be either master or replica" - break - elif k == 'postgres_version': - if not re.match(r'([1-9][0-9]?\.){2}[1-9][0-9]?$', request[k]): - status_code = 400 - data = "PostgreSQL version should be in the first.major.minor format" - break - elif k != 'with_pending_restart_flag': - status_code = 400 - data = "Unknown filter for the scheduled restart: {0}".format(k) + request = request or {} + if request: + logger.debug("received restart request: {0}".format(request)) + for k in request: + if k == 'schedule': + (_, data, request[k]) = self.parse_schedule(request[k], "restart") + if _: + status_code = _ break - else: - if 'schedule' not in request: + elif k == 'role': + if request[k] not in ('master', 'replica'): + status_code = 400 + data = "PostgreSQL role should be either master or replica" + break + elif k == 'postgres_version': + if not re.match(r'([1-9][0-9]?\.){2}[1-9][0-9]?$', request[k]): + status_code = 400 + data = "PostgreSQL version should be in the first.major.minor format" + break + elif k != 'with_pending_restart_flag': + status_code = 400 + data = "Unknown filter for the scheduled restart: {0}".format(k) + break + else: + if 'schedule' not in request: + try: + status, data = self.server.patroni.ha.restart() + status_code = 200 if status else 503 + except Exception: + logger.exception('Exception during restart') data = "Schedule required for the scheduled restart" status_code = 400 + else: + request['postmaster_start_time'] = self.server.patroni.ha.state_handler.postmaster_start_time() + if self.server.patroni.ha.schedule_future_restart(request): + data = "Restart scheduled" + status_code = 202 else: - request['postmaster_start_time'] = self.server.patroni.ha.state_handler.postmaster_start_time() - if self.server.patroni.ha.schedule_future_restart(request): - data = "Restart scheduled" - status_code = 202 - else: - data = "Another restart is already scheduled" - status_code = 409 + data = "Another restart is already scheduled" + status_code = 409 self._write_response(status_code, data) @check_auth def do_DELETE_restart(self): - self.server.patroni.ha.delete_future_restart(take_lock=True) + self.server.patroni.ha.delete_future_restart() data = "scheduled restart deleted" code = 200 self._write_response(code, data) diff --git a/patroni/async_executor.py b/patroni/async_executor.py index e009ab19..640ef992 100644 --- a/patroni/async_executor.py +++ b/patroni/async_executor.py @@ -1,5 +1,5 @@ import logging -from threading import Lock, Thread +from threading import RLock, Thread logger = logging.getLogger(__name__) @@ -8,9 +8,9 @@ class AsyncExecutor(object): def __init__(self): self._busy = False - self._thread_lock = Lock() + self._thread_lock = RLock() self._scheduled_action = None - self._scheduled_action_lock = Lock() + self._scheduled_action_lock = RLock() @property def busy(self): @@ -32,6 +32,7 @@ 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: diff --git a/patroni/ha.py b/patroni/ha.py index 6ac09533..b24bef94 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -401,17 +401,15 @@ class Ha(object): if (restart_data and self.should_run_scheduled_action('restart', restart_data['schedule'], self.delete_future_restart)): try: - if self.scheduled_restart_matches(restart_data.get('role'), - restart_data.get('postgres_version'), - ('with_pending_restart_flag' in restart_data)): - if self.state_handler.restart(): + ret, message = self.restart(restart_data) + if ret: logger.info("Scheduled restart successfull") else: - logger.warning("Scheduled restart failed") + logger.warning("Scheduled restart: {0}".format(message)) finally: self.delete_future_restart() - def scheduled_restart_matches(self, role, postgres_version, pending_restart): + def restart_matches(self, role, postgres_version, pending_restart): reason_to_cancel = "" # checking the restart filters here seem to be less ugly than moving them into the # run_scheduled_action. @@ -428,7 +426,7 @@ class Ha(object): if not reason_to_cancel: return True else: - logger.info("not proceeding with the scheduled restart: {0}".format(reason_to_cancel)) + logger.info("not proceeding with the restart: {0}".format(reason_to_cancel)) return False def schedule(self, action): @@ -443,11 +441,8 @@ class Ha(object): return True return False - def delete_future_restart(self, take_lock=False): - if take_lock: - with self._async_executor: - self.patroni.scheduled_restart = {} - else: + def delete_future_restart(self): + with self._async_executor: self.patroni.scheduled_restart = {} def immediate_restart_scheduled(self): @@ -463,7 +458,14 @@ class Ha(object): def reinitialize_scheduled(self): return self._async_executor.scheduled_action == 'reinitialize' - def restart(self): + def restart(self, restart_data=None): + """ conditional and unconditional restart """ + if (restart_data and isinstance(restart_data, dict) and + not self.restart_matches(restart_data.get('role'), + restart_data.get('postgres_version'), + ('with_pending_restart_flag' in restart_data))): + return (False, "restart conditions are not satisfied") + with self._async_executor: prev = self._async_executor.schedule('restart', True) if prev is not None: