synchronous_standby_names must be quoted with quote_ident (#505)

in addition to that implement additional checks around manual failover and recover when synchronous_mode is enabled

* Comparison must be case insensitive

patroni/api.py

@@ -293,9 +293,15 @@ class RestApiHandler(BaseHTTPRequestHandler):
         if leader and (not cluster.leader or cluster.leader.name != leader):
             return 'leader name does not match'
         if candidate:
+            if cluster.is_synchronous_mode() and cluster.sync.sync_standby != candidate:
+                return 'candidate name does not match with sync_standby'
             members = [m for m in cluster.members if m.name == candidate]
             if not members:
                 return 'candidate does not exists'
+        elif cluster.is_synchronous_mode():
+            members = [m for m in cluster.members if m.name == cluster.sync.sync_standby]
+            if not members:
+                return 'failover is not possible: can not find sync_standby'
         else:
             members = [m for m in cluster.members if m.name != cluster.leader.name and m.api_url]
             if not members:

patroni/dcs/__init__.py

@@ -320,6 +320,12 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat
     def is_paused(self):
         return self.config and self.config.data.get('pause', False) or False
 
+    def is_synchronous_mode(self):
+        return bool(self.config and self.config.data.get('synchronous_mode'))
+
+    def is_synchronous_mode_strict(self):
+        return bool(self.config and self.config.data.get('synchronous_mode_strict'))
+
 
 @six.add_metaclass(abc.ABCMeta)
 class AbstractDCS(object):

patroni/ha.py

@@ -184,7 +184,10 @@ class Ha(object):
             if timeout == 0:
                 # We are requested to prefer failing over to restarting master. But see first if there
                 # is anyone to fail over to.
-                if self.is_failover_possible(self.cluster.members):
+                members = self.cluster.members
+                if self.is_synchronous_mode():
+                    members = [m for m in members if self.cluster.sync.matches(m.name)]
+                if self.is_failover_possible(members):
                     logger.info("Master crashed. Failing over.")
                     self.demote('immediate')
                     return 'stopped PostgreSQL to fail over after a crash'
@@ -249,10 +252,10 @@ class Ha(object):
         return follow_reason
 
     def is_synchronous_mode(self):
-        return bool(self.cluster and self.cluster.config and self.cluster.config.data.get('synchronous_mode'))
+        return bool(self.cluster and self.cluster.is_synchronous_mode())
 
     def is_synchronous_mode_strict(self):
-        return bool(self.cluster and self.cluster.config and self.cluster.config.data.get('synchronous_mode_strict'))
+        return bool(self.cluster and self.cluster.is_synchronous_mode_strict())
 
     def process_sync_replication(self):
         """Process synchronous standby beahvior.
@@ -548,8 +551,8 @@ class Ha(object):
         self.state_handler.set_role('demoted')
 
         if mode_control['release']:
-                self.release_leader_key_voluntarily()
-                time.sleep(2)  # Give a time to somebody to take the leader lock
+            self.release_leader_key_voluntarily()
+            time.sleep(2)  # Give a time to somebody to take the leader lock
         if mode_control['offline']:
             node_to_follow, leader = None, None
         else:
@@ -563,6 +566,8 @@ class Ha(object):
             self._async_executor.schedule('starting after demotion')
             self._async_executor.run_async(self.state_handler.follow, (node_to_follow,))
         else:
+            if self.is_synchronous_mode():
+                self.state_handler.set_synchronous_standby(None)
             if self.state_handler.rewind_needed_and_possible(leader):
                 return False  # do not start postgres, but run pg_rewind on the next iteration
             self.state_handler.follow(node_to_follow)
@@ -621,8 +626,16 @@ class Ha(object):
                 if not failover.candidate and self.is_paused():
                     logger.warning('Failover is possible only to a specific candidate in a paused state')
                 else:
-                    members = [m for m in self.cluster.members
-                               if not failover.candidate or m.name == failover.candidate]
+                    if self.is_synchronous_mode():
+                        if failover.candidate and not self.cluster.sync.matches(failover.candidate):
+                            logger.warning('Failover candidate=%s does not match with sync_standby=%s',
+                                           failover.candidate, self.cluster.sync.sync_standby)
+                            members = []
+                        else:
+                            members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)]
+                    else:
+                        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, ('graceful',))

patroni/postgresql.py

@@ -17,7 +17,7 @@ from contextlib import contextmanager
 from patroni import call_self
 from patroni.callback_executor import CallbackExecutor
 from patroni.exceptions import PostgresConnectionException
-from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop, null_context
+from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop
 from six import string_types
 from six.moves.urllib.parse import quote_plus
 from threading import current_thread, Lock
@@ -42,6 +42,12 @@ STOP_SIGNALS = {
 }
 STOP_POLLING_INTERVAL = 1
 REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECK': 1, 'NEED': 2, 'NOT_NEED': 3, 'SUCCESS': 4, 'FAILED': 5})
+sync_standby_name_re = re.compile('^[A-Za-z_][A-Za-z_0-9\$]*$')
+
+
+def quote_ident(value):
+    """Very simplified version of quote_ident"""
+    return value if sync_standby_name_re.match(value) else '"' + value + '"'
 
 
 def slot_name_from_member_name(member_name):
@@ -60,6 +66,11 @@ def slot_name_from_member_name(member_name):
     return slot_name[0:63]
 
 
+@contextmanager
+def null_context():
+    yield
+
+
 class Postgresql(object):
 
     # List of parameters which must be always passed to postmaster as command line options
@@ -1652,12 +1663,13 @@ $$""".format(name, ' '.join(options)), name, password, password)
         :returns tuple of candidate name or None, and bool showing if the member is the active synchronous standby.
         """
         current = cluster.sync.sync_standby
-        members = {m.name: m for m in cluster.members}
+        current = current.lower() if current else current
+        members = {m.name.lower(): m for m in cluster.members}
         candidates = []
         # Pick candidates based on who has flushed WAL farthest.
         # TODO: for synchronous_commit = remote_write we actually want to order on write_location
         for app_name, state, sync_state in self.query(
-                """SELECT application_name, state, sync_state
+                """SELECT LOWER(application_name), state, sync_state
                      FROM pg_stat_replication
                     ORDER BY flush_{0} DESC""".format(self.lsn_name)):
             member = members.get(app_name)
@@ -1677,14 +1689,17 @@ $$""".format(name, ' '.join(options)), name, password, password)
 
     def set_synchronous_standby(self, name):
         """Sets a node to be synchronous standby and if changed does a reload for PostgreSQL."""
+        if name and name != '*':
+            name = quote_ident(name)
         if name != self._synchronous_standby_names:
             if name is None:
                 self._server_parameters.pop('synchronous_standby_names', None)
             else:
                 self._server_parameters['synchronous_standby_names'] = name
             self._synchronous_standby_names = name
-            self._write_postgresql_conf()
-            self.reload()
+            if self.state == 'running':
+                self._write_postgresql_conf()
+                self.reload()
 
     @staticmethod
     def postgres_version_to_int(pg_version):

patroni/utils.py

@@ -1,4 +1,3 @@
-import contextlib
 import random
 import time
 import re
@@ -281,8 +280,3 @@ def polling_loop(timeout, interval=1):
         yield iteration
         iteration += 1
         time.sleep(interval)
-
-
-@contextlib.contextmanager
-def null_context():
-    yield

tests/test_api.py

@@ -281,6 +281,7 @@ class TestRestApiHandler(unittest.TestCase):
     def test_do_POST_failover(self, dcs):
         dcs.loop_wait = 10
         cluster = dcs.get_cluster.return_value
+        cluster.is_synchronous_mode.return_value = False
 
         post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: '
 
@@ -292,14 +293,16 @@ class TestRestApiHandler(unittest.TestCase):
         cluster.leader.name = 'postgresql1'
         MockRestApiServer(RestApiHandler, request)
 
-        MockRestApiServer(RestApiHandler, post + '25\n\n{"leader": "postgresql1"}')
+        for cluster.is_synchronous_mode.return_value in (True, False):
+            MockRestApiServer(RestApiHandler, post + '25\n\n{"leader": "postgresql1"}')
 
         cluster.leader.name = 'postgresql2'
         request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
         MockRestApiServer(RestApiHandler, request)
 
         cluster.leader.name = 'postgresql1'
-        MockRestApiServer(RestApiHandler, request)
+        for cluster.is_synchronous_mode.return_value in (True, False):
+            MockRestApiServer(RestApiHandler, request)
 
         cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}),
                            Member(0, 'postgresql2', 30, {'api_url': 'http'})]

tests/test_ha.py

@@ -158,7 +158,6 @@ class TestHa(unittest.TestCase):
             self.ha.old_cluster = self.e.get_cluster()
             self.ha.cluster = get_cluster_not_initialized_without_leader()
             self.ha.load_cluster_from_dcs = Mock()
-            self.ha.is_synchronous_mode = false
 
     def test_update_lock(self):
         self.p.last_operation = Mock(side_effect=PostgresConnectionException(''))
@@ -438,6 +437,19 @@ class TestHa(unittest.TestCase):
         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_from_leader_in_synchronous_mode(self):
+        self.p.is_leader = true
+        self.ha.has_lock = true
+        self.ha.is_synchronous_mode = true
+        self.ha.is_failover_possible = false
+        self.ha.process_sync_replication = Mock()
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, None))
+        self.assertEquals('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, 'a', None), (self.p.name, 'a'))
+        self.ha.is_failover_possible = true
+        self.assertEquals('manual failover: demoting myself', self.ha.run_cycle())
+
     @patch('requests.get', requests_get)
     def test_manual_failover_process_no_leader(self):
         self.p.is_leader = false
@@ -635,7 +647,8 @@ class TestHa(unittest.TestCase):
     @patch('patroni.ha.Ha.demote')
     def test_failover_immediately_on_zero_master_start_timeout(self, demote):
         self.p.is_running = false
-        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.ha.cluster = get_cluster_initialized_with_leader(sync=(self.p.name, 'other'))
+        self.ha.cluster.config.data['synchronous_mode'] = True
         self.ha.patroni.config.set_dynamic_configuration({'master_start_timeout': 0})
         self.ha.has_lock = true
         self.ha.update_lock = true