Restart the node according to the schedule.

The scheduled restart data structures are now independent of those
used by the normal restarts. This would be fixed in subsequent
commits.
Add the behave tests, that cover the POST /restart (but not DELETE).

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
+

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)

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)

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

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

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