Implement postgresql-10 support (#444)

Mainly it handles rename of xlog to wal.
In the API and inside DCS it is still named xlog (for compatibility).

* Address feedback

patroni/api.py

@@ -382,14 +382,16 @@ class RestApiHandler(BaseHTTPRequestHandler):
                                        pg_is_in_recovery(),
                                        CASE WHEN pg_is_in_recovery()
                                             THEN 0
-                                            ELSE pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint
+                                            ELSE pg_{0}_{1}_diff(pg_current_{0}_{1}(), '0/0')::bigint
                                        END,
-                                       pg_xlog_location_diff(COALESCE(pg_last_xlog_receive_location(),
-                                                                      pg_last_xlog_replay_location()), '0/0')::bigint,
-                                       pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint,
+                                       pg_{0}_{1}_diff(COALESCE(pg_last_{0}_receive_{1}(),
+                                                                      pg_last_{0}_replay_{1}()), '0/0')::bigint,
+                                       pg_{0}_{1}_diff(pg_last_{0}_replay_{1}(), '0/0')::bigint,
                                        to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),
-                                       pg_is_in_recovery() AND pg_is_xlog_replay_paused(),
-                                       (SELECT array_to_json(array_agg(row_to_json(ri))) FROM replication_info ri)""",
+                                       pg_is_in_recovery() AND pg_is_{0}_replay_paused(),
+                                       (SELECT array_to_json(array_agg(row_to_json(ri)))
+                                          FROM replication_info ri)""".format(self.server.patroni.postgresql.wal_name,
+                                                                              self.server.patroni.postgresql.lsn_name),
                              retry=retry)[0]
 
             result = {

patroni/ha.py

@@ -18,20 +18,20 @@ from threading import RLock
 logger = logging.getLogger(__name__)
 
 
-class _MemberStatus(namedtuple('_MemberStatus', 'member,reachable,in_recovery,xlog_location,tags')):
+class _MemberStatus(namedtuple('_MemberStatus', 'member,reachable,in_recovery,wal_position,tags')):
     """Node status distilled from API response:
 
         member - dcs.Member object of the node
         reachable - `!False` if the node is not reachable or is not responding with correct JSON
         in_recovery - `!True` if pg_is_in_recovery() == true
-        xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role.
+        wal_position - value of `replayed_location` or `location` from JSON, dependin on its role.
         tags - dictionary with values of different tags (i.e. nofailover)
     """
     @classmethod
     def from_api_response(cls, member, json):
         is_master = json['role'] == 'master'
-        xlog = not is_master and max(json['xlog'].get('received_location', 0), json['xlog'].get('replayed_location', 0))
-        return cls(member, True, not is_master, xlog, json.get('tags', {}))
+        wal = not is_master and max(json['xlog'].get('received_location', 0), json['xlog'].get('replayed_location', 0))
+        return cls(member, True, not is_master, wal, json.get('tags', {}))
 
     @classmethod
     def unknown(cls, member):
@@ -116,7 +116,7 @@ class Ha(object):
                 data['pending_restart'] = True
             if not self._async_executor.busy and data['state'] in ['running', 'restarting', 'starting']:
                 try:
-                    data['xlog_location'] = self.state_handler.xlog_position(retry=False)
+                    data['xlog_location'] = self.state_handler.wal_position(retry=False)
                 except:
                     pass
             if self.patroni.scheduled_restart:
@@ -378,21 +378,21 @@ class Ha(object):
         pool.join()
         return results
 
-    def is_lagging(self, xlog_location):
-        """Returns if instance with an xlog should consider itself unhealthy to be promoted due to replication lag.
+    def is_lagging(self, wal_position):
+        """Returns if instance with an wal should consider itself unhealthy to be promoted due to replication lag.
 
-        :param xlog_location: Current xlog location.
+        :param wal_position: Current wal position.
         :returns True when node is lagging
         """
-        lag = (self.cluster.last_leader_operation or 0) - xlog_location
+        lag = (self.cluster.last_leader_operation or 0) - wal_position
         return lag > self.state_handler.config.get('maximum_lag_on_failover', 0)
 
     def _is_healthiest_node(self, members, check_replication_lag=True):
         """This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
 
-        my_xlog_location = self.state_handler.xlog_position()
-        if check_replication_lag and self.is_lagging(my_xlog_location):
-            return False  # Too far behind last reported xlog location on master
+        my_wal_position = self.state_handler.wal_position()
+        if check_replication_lag and self.is_lagging(my_wal_position):
+            return False  # Too far behind last reported wal position on master
 
         # Prepare list of nodes to run check against
         members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url]
@@ -403,7 +403,7 @@ class Ha(object):
                     if not st.in_recovery:
                         logger.warning('Master (%s) is still alive', st.member.name)
                         return False
-                    if my_xlog_location < st.xlog_location:
+                    if my_wal_position < st.wal_position:
                         return False
         return True
 
@@ -415,7 +415,7 @@ class Ha(object):
                 not_allowed_reason = st.failover_limitation()
                 if not_allowed_reason:
                     logger.info('Member %s is %s', st.member.name, not_allowed_reason)
-                elif self.is_lagging(st.xlog_location):
+                elif self.is_lagging(st.wal_position):
                     logger.info('Member %s exceeds maximum replication lag', st.member.name)
                 else:
                     ret = True

patroni/postgresql.py

@@ -66,20 +66,20 @@ class Postgresql(object):
     #    check_function -- if the new value is not correct must return `!False`
     #    min_version -- major version of PostgreSQL when parameter was introduced
     CMDLINE_OPTIONS = {
-        'listen_addresses': (None, lambda _: False, 9.1),
-        'port': (None, lambda _: False, 9.1),
-        'cluster_name': (None, lambda _: False, 9.5),
-        'wal_level': ('hot_standby', lambda v: v.lower() in ('hot_standby', 'replica', 'logical'), 9.1),
-        'hot_standby': ('on', lambda _: False, 9.1),
-        'max_connections': (100, lambda v: int(v) >= 100, 9.1),
-        'max_wal_senders': (5, lambda v: int(v) >= 5, 9.1),
-        'wal_keep_segments': (8, lambda v: int(v) >= 8, 9.1),
-        'max_prepared_transactions': (0, lambda v: int(v) >= 0, 9.1),
-        'max_locks_per_transaction': (64, lambda v: int(v) >= 64, 9.1),
-        'track_commit_timestamp': ('off', lambda v: parse_bool(v) is not None, 9.5),
-        'max_replication_slots': (5, lambda v: int(v) >= 5, 9.4),
-        'max_worker_processes': (8, lambda v: int(v) >= 8, 9.4),
-        'wal_log_hints': ('on', lambda _: False, 9.4)
+        'listen_addresses': (None, lambda _: False, 90100),
+        'port': (None, lambda _: False, 90100),
+        'cluster_name': (None, lambda _: False, 90500),
+        'wal_level': ('hot_standby', lambda v: v.lower() in ('hot_standby', 'replica', 'logical'), 90100),
+        'hot_standby': ('on', lambda _: False, 90100),
+        'max_connections': (100, lambda v: int(v) >= 100, 90100),
+        'max_wal_senders': (5, lambda v: int(v) >= 5, 90100),
+        'wal_keep_segments': (8, lambda v: int(v) >= 8, 90100),
+        'max_prepared_transactions': (0, lambda v: int(v) >= 0, 90100),
+        'max_locks_per_transaction': (64, lambda v: int(v) >= 64, 90100),
+        'track_commit_timestamp': ('off', lambda v: parse_bool(v) is not None, 90500),
+        'max_replication_slots': (5, lambda v: int(v) >= 5, 90400),
+        'max_worker_processes': (8, lambda v: int(v) >= 8, 90400),
+        'wal_log_hints': ('on', lambda _: False, 90400)
     }
 
     def __init__(self, config):
@@ -153,7 +153,7 @@ class Postgresql(object):
 
     @property
     def use_slots(self):
-        return self._use_slots and self._major_version >= 9.4
+        return self._use_slots and self._major_version >= 90400
 
     @property
     def _replication(self):
@@ -163,6 +163,18 @@ class Postgresql(object):
     def callback(self):
         return self.config.get('callbacks') or {}
 
+    @staticmethod
+    def _wal_name(version):
+        return 'wal' if version >= 100000 else 'xlog'
+
+    @property
+    def wal_name(self):
+        return self._wal_name(self._major_version)
+
+    @property
+    def lsn_name(self):
+        return 'lsn' if self._major_version >= 100000 else 'location'
+
     def _version_file_exists(self):
         return not self.data_directory_empty() and os.path.isfile(self._version_file)
 
@@ -170,10 +182,10 @@ class Postgresql(object):
         if self._version_file_exists():
             try:
                 with open(self._version_file) as f:
-                    return float(f.read())
+                    return self.postgres_major_version_to_int(f.read().strip())
             except Exception:
                 logger.exception('Failed to read PG_VERSION from %s', self._data_dir)
-        return 0.0
+        return 0
 
     def get_server_parameters(self, config):
         parameters = config['parameters'].copy()
@@ -187,10 +199,10 @@ class Postgresql(object):
                     parameters.pop('synchronous_standby_names', None)
             else:
                 parameters['synchronous_standby_names'] = self._synchronous_standby_names
-        if self._major_version >= 9.6 and parameters['wal_level'] == 'hot_standby':
+        if self._major_version >= 90600 and parameters['wal_level'] == 'hot_standby':
             parameters['wal_level'] = 'replica'
         return {k: v for k, v in parameters.items() if not self._major_version or
-                self._major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 9.1))[2]}
+                self._major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 90100))[2]}
 
     def resolve_connection_addresses(self):
         self._local_address = self.get_local_address()
@@ -1165,13 +1177,13 @@ BEGIN
 END;
 $$""".format(name, ' '.join(options)), name, password, password)
 
-    def xlog_position(self, retry=True):
+    def wal_position(self, retry=True):
         stmt = """SELECT CASE WHEN pg_is_in_recovery()
-                              THEN GREATEST(pg_xlog_location_diff(COALESCE(pg_last_xlog_receive_location(), '0/0'),
+                              THEN GREATEST(pg_{0}_{1}_diff(COALESCE(pg_last_{0}_receive_{1}(), '0/0'),
                                                                   '0/0')::bigint,
-                                            pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint)
-                              ELSE pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint
-                          END"""
+                                            pg_{0}_{1}_diff(pg_last_{0}_replay_{1}(), '0/0')::bigint)
+                              ELSE pg_{0}_{1}_diff(pg_current_{0}_{1}(), '0/0')::bigint
+                          END""".format(self.wal_name, self.lsn_name)
 
         # This method could be called from different threads (simultaneously with some other `_query` calls).
         # If it is called not from main thread we will create a new cursor to execute statement.
@@ -1243,7 +1255,7 @@ $$""".format(name, ' '.join(options)), name, password, password)
                 self._schedule_load_slots = True
 
     def last_operation(self):
-        return str(self.xlog_position())
+        return str(self.wal_position())
 
     def clone(self, clone_member):
         """
@@ -1297,6 +1309,11 @@ $$""".format(name, ' '.join(options)), name, password, password)
             self.move_data_directory()
 
     def basebackup(self, conn_url, env):
+        # save environ to restore it later
+        old_env = os.environ.copy()
+        os.environ.clear()
+        os.environ.update(env)
+
         # creates a replica data dir using pg_basebackup.
         # this is the default, built-in create_replica_method
         # tries twice, then returns failure (as 1)
@@ -1308,13 +1325,19 @@ $$""".format(name, ' '.join(options)), name, password, password)
                 self.remove_data_directory()
 
             try:
+                version = 0
+                with psycopg2.connect(conn_url + '?replication=1') as c:
+                    version = c.server_version
+
                 ret = subprocess.call([self._pgcommand('pg_basebackup'), '--pgdata=' + self._data_dir,
-                                       '--xlog-method=stream', "--dbname=" + conn_url], env=env)
+                                       '--{0}-method=stream'.format(self._wal_name(version)), '--dbname=' + conn_url])
                 if ret == 0:
                     break
                 else:
                     logger.error('Error when fetching backup: pg_basebackup exited with code=%s', ret)
 
+            except psycopg2.Error:
+                logger.error('Can not connect to %s', conn_url)
             except Exception as e:
                 logger.error('Error when fetching backup with pg_basebackup: %s', e)
 
@@ -1322,6 +1345,10 @@ $$""".format(name, ' '.join(options)), name, password, password)
                 logger.warning('Trying again in 5 seconds')
                 time.sleep(5)
 
+        # restore environ
+        os.environ.clear()
+        os.environ.update(old_env)
+
         return ret
 
     def pick_synchronous_standby(self, cluster):
@@ -1340,7 +1367,7 @@ $$""".format(name, ' '.join(options)), name, password, password)
         for app_name, state, sync_state in self.query(
                 """SELECT application_name, state, sync_state
                      FROM pg_stat_replication
-                    ORDER BY flush_location DESC"""):
+                    ORDER BY flush_{0} DESC""".format(self.lsn_name)):
             member = members.get(app_name)
             if state != 'streaming' or not member or member.tags.get('nosync', False):
                 continue
@@ -1400,3 +1427,13 @@ $$""".format(name, ' '.join(options)), name, password, password)
         except ValueError:
             raise Exception("Invalid PostgreSQL version: {0}".format(pg_version))
         return result
+
+    @staticmethod
+    def postgres_major_version_to_int(pg_version):
+        """
+        >>> Postgresql.postgres_major_version_to_int('10')
+        100000
+        >>> Postgresql.postgres_major_version_to_int('9.6')
+        90600
+        """
+        return Postgresql.postgres_version_to_int(pg_version + '.0')

tests/test_api.py

@@ -25,6 +25,8 @@ class MockPostgresql(object):
     sysid = 'dummysysid'
     scope = 'dummy'
     pending_restart = True
+    wal_name = 'wal'
+    lsn_name = 'lsn'
 
     @staticmethod
     def connection():
@@ -64,7 +66,7 @@ class MockHa(object):
         return True
 
     @staticmethod
-    def is_lagging(xlog):
+    def is_lagging(wal):
         return False
 
     @staticmethod

tests/test_ha.py

@@ -54,12 +54,12 @@ def get_cluster_initialized_with_only_leader(failover=None):
     return get_cluster(True, l, [l], failover, None)
 
 
-def get_node_status(reachable=True, in_recovery=True, xlog_location=10, nofailover=False):
+def get_node_status(reachable=True, in_recovery=True, wal_position=10, nofailover=False):
     def fetch_node_status(e):
         tags = {}
         if nofailover:
             tags['nofailover'] = True
-        return _MemberStatus(e, reachable, in_recovery, xlog_location, tags)
+        return _MemberStatus(e, reachable, in_recovery, wal_position, tags)
     return fetch_node_status
 
 future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5)
@@ -109,13 +109,13 @@ def run_async(self, func, args=()):
 
 @patch.object(Postgresql, 'is_running', Mock(return_value=True))
 @patch.object(Postgresql, 'is_leader', Mock(return_value=True))
-@patch.object(Postgresql, 'xlog_position', Mock(return_value=10))
+@patch.object(Postgresql, 'wal_position', Mock(return_value=10))
 @patch.object(Postgresql, 'call_nowait', Mock(return_value=True))
 @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
 @patch.object(Postgresql, 'controldata', Mock(return_value={'Database system identifier': '1234567890'}))
 @patch.object(Postgresql, 'sync_replication_slots', Mock())
 @patch.object(Postgresql, 'write_pg_hba', Mock())
-@patch.object(Postgresql, 'write_pgpass', Mock())
+@patch.object(Postgresql, 'write_pgpass', Mock(return_value={}))
 @patch.object(Postgresql, 'write_recovery_conf', Mock())
 @patch.object(Postgresql, 'query', Mock())
 @patch.object(Postgresql, 'checkpoint', Mock())
@@ -159,7 +159,7 @@ class TestHa(unittest.TestCase):
         self.assertTrue(self.ha.update_lock(True))
 
     def test_touch_member(self):
-        self.p.xlog_position = Mock(side_effect=Exception)
+        self.p.wal_position = Mock(side_effect=Exception)
         self.ha.touch_member()
 
     def test_start_as_replica(self):
@@ -276,6 +276,7 @@ class TestHa(unittest.TestCase):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
         self.assertEquals(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader')
 
+    @patch('time.sleep', Mock())
     def test_bootstrap_from_another_member(self):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from replica \'other\'')
@@ -304,6 +305,7 @@ class TestHa(unittest.TestCase):
         self.p.bootstrap = Mock(side_effect=PostgresException("Could not bootstrap master PostgreSQL"))
         self.assertRaises(PostgresException, self.ha.bootstrap)
 
+    @patch('psycopg2.connect', psycopg2_connect)
     def test_reinitialize(self):
         self.assertIsNotNone(self.ha.reinitialize())
 
@@ -315,6 +317,7 @@ class TestHa(unittest.TestCase):
         self.ha.state_handler.name = self.ha.cluster.leader.name
         self.assertIsNotNone(self.ha.reinitialize())
 
+    @patch('time.sleep', Mock())
     def test_restart(self):
         self.assertEquals(self.ha.restart({}), (True, 'restarted successfully'))
         self.p.restart = Mock(return_value=None)
@@ -359,7 +362,7 @@ class TestHa(unittest.TestCase):
         self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself')
         self.ha.fetch_node_status = get_node_status(nofailover=True)
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
-        self.ha.fetch_node_status = get_node_status(xlog_location=1)
+        self.ha.fetch_node_status = get_node_status(wal_position=1)
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
         # manual failover from the previous leader to us won't happen if we hold the nofailover flag
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None))
@@ -451,9 +454,9 @@ class TestHa(unittest.TestCase):
         self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.ha.fetch_node_status = get_node_status(in_recovery=False)  # accessible, not in_recovery
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        self.ha.fetch_node_status = get_node_status(xlog_location=11)  # accessible, in_recovery, xlog location ahead
+        self.ha.fetch_node_status = get_node_status(wal_position=11)  # accessible, in_recovery, wal position ahead
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        with patch('patroni.postgresql.Postgresql.xlog_position', return_value=1):
+        with patch('patroni.postgresql.Postgresql.wal_position', return_value=1):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.ha.patroni.nofailover = True
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
@@ -786,6 +789,7 @@ class TestHa(unittest.TestCase):
     def test_wakup(self):
         self.ha.wakeup()
 
+    @patch('time.sleep', Mock())
     def test_leader_with_empty_directory(self):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.ha.has_lock = true

tests/test_postgresql.py

@@ -72,7 +72,7 @@ class MockCursor(object):
 
 class MockConnect(object):
 
-    server_version = '99999'
+    server_version = 99999
     autocommit = False
     closed = 0
 
@@ -161,7 +161,7 @@ class TestPostgresql(unittest.TestCase):
     @patch('subprocess.call', Mock(return_value=0))
     @patch('psycopg2.connect', psycopg2_connect)
     @patch('os.rename', Mock())
-    @patch.object(Postgresql, 'get_major_version', Mock(return_value=9.6))
+    @patch.object(Postgresql, 'get_major_version', Mock(return_value=90600))
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def setUp(self):
         self.data_dir = 'data/test0'
@@ -599,9 +599,9 @@ class TestPostgresql(unittest.TestCase):
     @patch.object(Postgresql, '_version_file_exists', Mock(return_value=True))
     def test_get_major_version(self):
         with patch.object(builtins, 'open', mock_open(read_data='9.4')):
-            self.assertEquals(self.p.get_major_version(), 9.4)
+            self.assertEquals(self.p.get_major_version(), 90400)
         with patch.object(builtins, 'open', Mock(side_effect=Exception)):
-            self.assertEquals(self.p.get_major_version(), 0.0)
+            self.assertEquals(self.p.get_major_version(), 0)
 
     def test_postmaster_start_time(self):
         with patch.object(MockCursor, "fetchone", Mock(return_value=('foo', True, '', '', '', '', False))):