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

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

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:

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: