From cd84dc82b6e3b747ab3300f0d9958113cff55f82 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 19 May 2017 17:04:53 +0200 Subject: [PATCH] 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 | 14 ++++--- patroni/ha.py | 28 ++++++------- patroni/postgresql.py | 91 ++++++++++++++++++++++++++++------------ tests/test_api.py | 4 +- tests/test_ha.py | 20 +++++---- tests/test_postgresql.py | 8 ++-- 6 files changed, 105 insertions(+), 60 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 56cbd0a4..e8381e8b 100644 --- a/patroni/api.py +++ b/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 = { diff --git a/patroni/ha.py b/patroni/ha.py index 460c0dff..512d9595 100644 --- a/patroni/ha.py +++ b/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 diff --git a/patroni/postgresql.py b/patroni/postgresql.py index d0036f4b..5ea327f4 100644 --- a/patroni/postgresql.py +++ b/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') diff --git a/tests/test_api.py b/tests/test_api.py index 0a227368..c58afbb8 100644 --- a/tests/test_api.py +++ b/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 diff --git a/tests/test_ha.py b/tests/test_ha.py index 51d8eb3c..39601e15 100644 --- a/tests/test_ha.py +++ b/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 diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index f5b2c7b2..92b34e3f 100644 --- a/tests/test_postgresql.py +++ b/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))):