diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py index e4281692..1a4477ea 100644 --- a/.github/workflows/install_deps.py +++ b/.github/workflows/install_deps.py @@ -39,7 +39,7 @@ def install_packages(what): } packages['exhibitor'] = packages['zookeeper'] packages = packages.get(what, []) - ver = str({'etcd': '9.6', 'etcd3': '9.6', 'consul': 10, 'exhibitor': 11, 'kubernetes': 12, 'raft': 13}.get(what)) + ver = str({'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what)) subprocess.call(['sudo', 'apt-get', 'update', '-y']) return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev', 'wget'] + packages) diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py index e2db5e93..fbda3999 100644 --- a/.github/workflows/run_tests.py +++ b/.github/workflows/run_tests.py @@ -23,18 +23,19 @@ def main(): env = os.environ.copy() if sys.platform.startswith('linux'): - version = {'etcd': '9.6', 'etcd3': '9.6', 'consul': 10, 'exhibitor': 11, 'kubernetes': 12, 'raft': 13}.get(what) + version = {'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what) path = '/usr/lib/postgresql/{0}/bin:.'.format(version) unbuffer = ['timeout', '600', 'unbuffer'] + args = ['--tags=-skip'] if what == 'etcd' else [] else: path = os.path.abspath(os.path.join('pgsql', 'bin')) if sys.platform == 'darwin': path += ':.' - unbuffer = [] + args = unbuffer = [] env['PATH'] = path + os.pathsep + env['PATH'] env['DCS'] = what - ret = subprocess.call(unbuffer + [sys.executable, '-m', 'behave'], env=env) + ret = subprocess.call(unbuffer + [sys.executable, '-m', 'behave'] + args, env=env) if ret != 0: if subprocess.call('grep . features/output/*_failed/*postgres?.*', shell=True) != 0: diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 499623b9..cd951e68 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -34,7 +34,7 @@ Dynamic configuration is stored in the DCS (Distributed Configuration Store) and - **restore\_command**: command to restore WAL records from the remote master to standby leader, can be different from the list defined in :ref:`postgresql_settings` - **archive\_cleanup\_command**: cleanup command for standby leader - **recovery\_min\_apply\_delay**: how long to wait before actually apply WAL records on a standby leader -- **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Patroni will try to create slots before opening connections to the cluster. +- **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+. - **my_slot_name**: the name of replication slot. If the permanent slot name matches with the name of the current primary it will not be created. Everything else is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots. - **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``. - **database**: the database name where logical slots should be created. diff --git a/features/callback.py b/features/callback.py deleted file mode 100755 index 767be49f..00000000 --- a/features/callback.py +++ /dev/null @@ -1,17 +0,0 @@ -#!/usr/bin/env python -import os -import psycopg2 -import sys - - -if __name__ == '__main__': - if not (len(sys.argv) >= 3 and sys.argv[3] == "master"): - sys.exit(1) - - os.environ['PGPASSWORD'] = 'zalando' - connection = psycopg2.connect(host='127.0.0.1', port=sys.argv[1], user='postgres') - cursor = connection.cursor() - cursor.execute("SELECT slot_name FROM pg_replication_slots WHERE slot_type = 'logical'") - - with open("data/postgres0/label", "w") as label: - label.write(next(iter(cursor.fetchone()), "")) diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature index 2e24c299..bad8c51a 100644 --- a/features/standby_cluster.feature +++ b/features/standby_cluster.feature @@ -1,5 +1,5 @@ Feature: standby cluster - Scenario: check permanent logical slots are preserved on failover/switchover + Scenario: prepare the cluster with logical slots Given I start postgres1 Then postgres1 is a leader after 10 seconds And there is a non empty initialize key in DCS after 15 seconds @@ -10,15 +10,24 @@ Feature: standby cluster When I issue a PATCH request to http://127.0.0.1:8009/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}} Then I receive a response code 200 And I do a backup of postgres1 - When I start postgres0 with callback configured + When I start postgres0 Then "members/postgres0" key in DCS has state=running after 10 seconds And replication works from postgres1 to postgres0 after 15 seconds + + @skip + Scenario: check permanent logical slots are synced to the replica + Given I run patronictl.py restart batman postgres1 --force + Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds + When I add the table replicate_me to postgres1 + And I get all changes from logical slot test_logical on postgres1 + Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds + + Scenario: Detach exiting node from the cluster When I shut down postgres1 Then postgres0 is a leader after 10 seconds And "members/postgres0" key in DCS has role=master after 3 seconds When I issue a GET request to http://127.0.0.1:8008/ Then I receive a response code 200 - And there is a label with "test_logical" in postgres0 data directory Scenario: check replication of a single table in a standby cluster Given I start postgres1 in a standby cluster batman1 as a clone of postgres0 @@ -35,6 +44,7 @@ Feature: standby cluster When I start postgres2 in a cluster batman1 Then postgres2 role is the replica after 24 seconds And table foo is present on postgres2 after 20 seconds + And postgres1 does not have a logical replication slot named test_logical Scenario: check failover When I kill postgres1 diff --git a/features/steps/slots.py b/features/steps/slots.py index 8e87b064..f761cceb 100644 --- a/features/steps/slots.py +++ b/features/steps/slots.py @@ -1,5 +1,7 @@ +import time +import psycopg2 + from behave import step, then -import psycopg2 as pg @step('I create a logical replication slot {slot_name} on {pg_name:w} with the {plugin:w} plugin') @@ -8,7 +10,7 @@ def create_logical_replication_slot(context, slot_name, pg_name, plugin): output = context.pctl.query(pg_name, ("SELECT pg_create_logical_replication_slot('{0}', '{1}')," " current_database()").format(slot_name, plugin)) print(output.fetchone()) - except pg.Error as e: + except psycopg2.Error as e: print(e) assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin) @@ -22,7 +24,7 @@ def has_logical_replication_slot(context, pg_name, slot_name, plugin): assert row[0] == "logical", "Found replication slot named {0} but wasn't a logical slot".format(slot_name) assert row[1] == plugin, ("Found replication slot named {0} but was using plugin " "{1} rather than {2}").format(slot_name, row[1], plugin) - except pg.Error: + except psycopg2.Error: assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin) @@ -32,5 +34,27 @@ def does_not_have_logical_replication_slot(context, pg_name, slot_name): row = context.pctl.query(pg_name, ("SELECT 1 FROM pg_replication_slots" " WHERE slot_name = '{0}'").format(slot_name)).fetchone() assert not row, "Found unexpected replication slot named {0}".format(slot_name) - except pg.Error: + except psycopg2.Error: assert False, "Error looking for slot {0} on {1}".format(slot_name, pg_name) + + +@step('Logical slot {slot_name:w} is in sync between {pg_name1:w} and {pg_name2:w} after {time_limit:d} seconds') +def logical_slots_in_sync(context, slot_name, pg_name1, pg_name2, time_limit): + time_limit *= context.timeout_multiplier + max_time = time.time() + int(time_limit) + while time.time() < max_time: + try: + query = "SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = '{0}'".format(slot_name) + slot1 = context.pctl.query(pg_name1, query).fetchone() + slot2 = context.pctl.query(pg_name2, query).fetchone() + if slot1[0] == slot2[0]: + return + except Exception: + pass + time.sleep(1) + assert False, "Logical slot {0} is not in sync between {1} and {2}".format(slot_name, pg_name1, pg_name2) + + +@step('I get all changes from logical slot {slot_name:w} on {pg_name:w}') +def logical_slot_get_changes(context, slot_name, pg_name): + context.pctl.query(pg_name, "SELECT * FROM pg_logical_slot_get_changes('{0}', NULL, NULL)".format(slot_name)) diff --git a/features/steps/standby_cluster.py b/features/steps/standby_cluster.py index c04fd6f3..eb011dae 100644 --- a/features/steps/standby_cluster.py +++ b/features/steps/standby_cluster.py @@ -14,17 +14,6 @@ executable = sys.executable if os.name != 'nt' else sys.executable.replace('\\', callback = executable + " features/callback2.py " -@step('I start {name:w} with callback configured') -def start_patroni_with_callbacks(context, name): - return context.pctl.start(name, custom_config={ - "postgresql": { - "callbacks": { - "on_role_change": executable + " features/callback.py" - } - } - }) - - @step('I start {name:w} in a cluster {cluster_name:w}') def start_patroni(context, name, cluster_name): return context.pctl.start(name, custom_config={ @@ -55,7 +44,8 @@ def start_patroni_standby_cluster(context, name, cluster_name, name2): "port": port, "primary_slot_name": "pm_1", "create_replica_methods": ["backup_restore", "basebackup"] - } + }, + "postgresql": {"parameters": {"wal_level": "logical"}} } }, "postgresql": { diff --git a/patroni/api.py b/patroni/api.py index e02d5acf..7bc9fc63 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -97,7 +97,7 @@ class RestApiHandler(BaseHTTPRequestHandler): patroni = self.server.patroni cluster = patroni.dcs.cluster - leader_optime = cluster and cluster.last_leader_operation or 0 + leader_optime = cluster and cluster.last_lsn or 0 replayed_location = response.get('xlog', {}).get('replayed_location', 0) max_replica_lag = parse_int(self.path_query.get('lag', [sys.maxsize])[0], 'B') if max_replica_lag is None: diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index e5503f3e..10c2d5f8 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -13,12 +13,13 @@ import time from collections import defaultdict, namedtuple from copy import deepcopy -from patroni.exceptions import PatroniFatalException -from patroni.utils import parse_bool, uri from random import randint from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl from threading import Event, Lock +from ..exceptions import PatroniFatalException +from ..utils import deep_compare, parse_bool, uri + slot_name_re = re.compile('^[a-z0-9_]{1,63}$') logger = logging.getLogger(__name__) @@ -133,6 +134,8 @@ class Member(namedtuple('Member', 'index,name,session,data')): else: try: data = json.loads(data) + if not isinstance(data, dict): + data = {} except (TypeError, ValueError): data = {} return Member(index, name, session, data) @@ -206,6 +209,15 @@ class Member(namedtuple('Member', 'index,name,session,data')): def is_running(self): return self.state == 'running' + @property + def version(self): + version = self.data.get('version') + if version: + try: + return tuple(map(int, version.split('.'))) + except Exception: + logger.debug('Failed to parse Patroni version %s', version) + class RemoteMember(Member): """ Represents a remote master for a standby cluster @@ -259,14 +271,10 @@ class Leader(namedtuple('Leader', 'index,session,member')): """ >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote """ - version = self.data.get('version') - if version: - try: - # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false - if tuple(map(int, version.split('.'))) > (1, 5, 6): - return self.data['role'] == 'master' and 'checkpoint_after_promote' not in self.data - except Exception: - logger.debug('Failed to parse Patroni version %s', version) + version = self.member.version + # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false + if version and version > (1, 5, 6): + return self.data.get('role') == 'master' and 'checkpoint_after_promote' not in self.data class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')): @@ -432,19 +440,20 @@ class TimelineHistory(namedtuple('TimelineHistory', 'index,value,lines')): return TimelineHistory(index, value, lines) -class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operation,members,failover,sync,history')): +class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,failover,sync,history,slots')): """Immutable object (namedtuple) which represents PostgreSQL cluster. Consists of the following fields: :param initialize: shows whether this cluster has initialization key stored in DC or not. :param config: global dynamic configuration, reference to `ClusterConfig` object :param leader: `Leader` object which represents current leader of the cluster - :param last_leader_operation: int or long object containing position of last known leader operation. - This value is stored in `/optime/leader` key + :param last_lsn: int or long object containing position of last known leader LSN. + This value is stored in the `/status` key or `/optime/leader` (legacy) key :param members: list of Member object, all PostgreSQL cluster members including leader :param failover: reference to `Failover` object :param sync: reference to `SyncState` object, last observed synchronous replication state. :param history: reference to `TimelineHistory` object + :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int} """ def is_unlocked(self): @@ -470,22 +479,41 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat def is_synchronous_mode(self): return self.check_mode('synchronous_mode') - def get_replication_slots(self, my_name, role): + @property + def __permanent_slots(self): + return self.config and self.config.permanent_slots or {} + + @property + def __permanent_physical_slots(self): + return {name: value for name, value in self.__permanent_slots.items() + if not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'} + + @property + def __permanent_logical_slots(self): + return {name: value for name, value in self.__permanent_slots.items() if isinstance(value, dict) + and value.get('type', 'logical') == 'logical' and value.get('database') and value.get('plugin')} + + @property + def use_slots(self): + return self.config and self.config.data.get('postgresql', {}).get('use_slots', True) + + def get_replication_slots(self, my_name, role, nofailover, major_version, show_error=False): # if the replicatefrom tag is set on the member - we should not create the replication slot for it on # the current master, because that member would replicate from elsewhere. We still create the slot if # the replicatefrom destination member is currently not a member of the cluster (fallback to the # master), or if replicatefrom destination member happens to be the current master - use_slots = self.config and self.config.data.get('postgresql', {}).get('use_slots', True) + use_slots = self.use_slots if role in ('master', 'standby_leader'): slot_members = [m.name for m in self.members if use_slots and m.name != my_name and (m.replicatefrom is None or m.replicatefrom == my_name or not self.has_member(m.replicatefrom))] - permanent_slots = (self.config and self.config.permanent_slots or {}).copy() + permanent_slots = self.__permanent_slots if use_slots and \ + role == 'master' else self.__permanent_physical_slots else: # only manage slots for replicas that replicate from this one, except for the leader among them slot_members = [m.name for m in self.members if use_slots and m.replicatefrom == my_name and m.name != self.leader.name] - permanent_slots = {} + permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {} slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members} @@ -499,6 +527,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat for k, v in slot_conflicts.items() if len(v) > 1)) # "merge" replication slots for members with permanent_replication_slots + disabled_permanent_logical_slots = [] for name, value in permanent_slots.items(): if not slot_name_re.match(name): logger.error("Invalid permanent replication slot name '%s'", name) @@ -516,7 +545,9 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat slots[name] = value continue elif value['type'] == 'logical' and value.get('database') and value.get('plugin'): - if name in slots: + if major_version < 110000: + disabled_permanent_logical_slots.append(name) + elif name in slots: logger.error("Permanent logical replication slot {'%s': %s} is conflicting with" + " physical replication slot for cluster member", name, value) else: @@ -525,20 +556,53 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name]) + if disabled_permanent_logical_slots and show_error: + logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. " + "Following slots will not be created: %s.", disabled_permanent_logical_slots) + return slots - def has_permanent_logical_slots(self, name): - slots = self.get_replication_slots(name, 'master').values() + def has_permanent_logical_slots(self, my_name, nofailover, major_version=110000): + if major_version < 110000: + return False + slots = self.get_replication_slots(my_name, 'replica', nofailover, major_version).values() return any(v for v in slots if v.get("type") == "logical") + def should_enforce_hot_standby_feedback(self, my_name, nofailover, major_version): + """ + The hot_standby_feedback must be enabled if the current replica has logical slots + or it is working as a cascading replica for the other node that has logical slots. + """ + + if major_version < 110000: + return False + + if self.has_permanent_logical_slots(my_name, nofailover, major_version): + return True + + if self.use_slots: + members = [m for m in self.members if m.replicatefrom == my_name and m.name != self.leader.name] + return any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover, major_version) for m in members) + return False + + def get_my_slot_name_on_primary(self, my_name, replicatefrom): + """ + P <-- I <-- L + In case of cascading replication we have to check not our physical slot, + but slot of the replica that connects us to the primary. + """ + + m = self.get_member(replicatefrom, False) if replicatefrom else None + return self.get_my_slot_name_on_primary(m.name, m.replicatefrom) if m else slot_name_from_member_name(my_name) + @property def timeline(self): """ - >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0).timeline + >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0, 0).timeline 0 - >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]')).timeline + >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]'), 0).timeline 1 - >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]')).timeline + >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]'), 0).timeline 0 """ if self.history: @@ -551,6 +615,10 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operat return 1 return 0 + @property + def min_version(self): + return next(iter(sorted(filter(lambda v: v, [m.version for m in self.members])) + [None])) + @six.add_metaclass(abc.ABCMeta) class AbstractDCS(object): @@ -562,7 +630,8 @@ class AbstractDCS(object): _HISTORY = 'history' _MEMBERS = 'members/' _OPTIME = 'optime' - _LEADER_OPTIME = _OPTIME + '/' + _LEADER + _STATUS = 'status' # JSON, containts "leader_lsn" and confirmed_flush_lsn of logical "slots" on the leader + _LEADER_OPTIME = _OPTIME + '/' + _LEADER # legacy _SYNC = 'sync' def __init__(self, config): @@ -578,7 +647,8 @@ class AbstractDCS(object): self._cluster = None self._cluster_valid_till = 0 self._cluster_thread_lock = Lock() - self._last_leader_operation = '' + self._last_lsn = '' + self._last_status = {} self.event = Event() def client_path(self, path): @@ -612,6 +682,10 @@ class AbstractDCS(object): def history_path(self): return self.client_path(self._HISTORY) + @property + def status_path(self): + return self.client_path(self._STATUS) + @property def leader_optime_path(self): return self.client_path(self._LEADER_OPTIME) @@ -682,14 +756,30 @@ class AbstractDCS(object): self._cluster_valid_till = 0 @abc.abstractmethod - def _write_leader_optime(self, last_operation): - """write current xlog location into `/optime/leader` key in DCS - :param last_operation: absolute xlog location in bytes + def _write_leader_optime(self, last_lsn): + """write current WAL LSN into `/optime/leader` key in DCS + + :param last_lsn: absolute WAL LSN in bytes :returns: `!True` on success.""" - def write_leader_optime(self, last_operation): - if self._last_leader_operation != last_operation and self._write_leader_optime(last_operation): - self._last_leader_operation = last_operation + def write_leader_optime(self, last_lsn): + if self._last_lsn != last_lsn and self._write_leader_optime(last_lsn): + self._last_lsn = last_lsn + + @abc.abstractmethod + def _write_status(self, value): + """write current WAL LSN and confirmed_flush_lsn of permanent slots into the `/status` key in DCS + + :param value: status serialized in JSON forman + :returns: `!True` on success.""" + + def write_status(self, value): + if not deep_compare(self._last_status, value) and self._write_status(json.dumps(value, separators=(',', ':'))): + self._last_status = value + cluster = self.cluster + min_version = cluster and cluster.min_version + if min_version and min_version < (2, 0, 3): + self._write_leader_optime(str(value[self._OPTIME])) @abc.abstractmethod def _update_leader(self): @@ -701,16 +791,20 @@ class AbstractDCS(object): You have to use CAS (Compare And Swap) operation in order to update leader key, for example for etcd `prevValue` parameter must be used.""" - def update_leader(self, last_operation, access_is_restricted=False): + def update_leader(self, last_lsn, slots=None): """Update leader key (or session) ttl and optime/leader - :param last_operation: absolute xlog location in bytes + :param last_lsn: absolute WAL LSN in bytes + :param slots: dict with permanent slots confirmed_flush_lsn :returns: `!True` if leader key (or session) has been updated successfully. If not, `!False` must be returned and current instance would be demoted.""" ret = self._update_leader() - if ret and last_operation: - self.write_leader_optime(last_operation) + if ret and last_lsn: + status = {self._OPTIME: last_lsn} + if slots: + status['slots'] = slots + self.write_status(status) return ret @abc.abstractmethod @@ -779,13 +873,13 @@ class AbstractDCS(object): """Remove leader key from DCS. This method should remove leader key if current instance is the leader""" - def delete_leader(self, last_operation=None): + def delete_leader(self, last_lsn=None): """Update optime/leader and voluntarily remove leader key from DCS. This method should remove leader key if current instance is the leader. - :param last_operation: latest checkpoint location in bytes""" + :param last_lsn: latest checkpoint location in bytes""" - if last_operation: - self.write_leader_optime(last_operation) + if last_lsn: + self.write_leader_optime(last_lsn) return self._delete_leader() @abc.abstractmethod diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index a9acb08b..6c395252 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -337,9 +337,24 @@ class Consul(AbstractDCS): history = nodes.get(self._HISTORY) history = history and TimelineHistory.from_node(history['ModifyIndex'], history['Value']) - # get last leader operation - last_leader_operation = nodes.get(self._LEADER_OPTIME) - last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['Value']) + # get last known leader lsn and slots + status = nodes.get(self._STATUS) + if status: + try: + status = json.loads(status['Value']) + last_lsn = status.get(self._OPTIME) + slots = status.get('slots') + except Exception: + slots = last_lsn = None + else: + last_lsn = nodes.get(self._LEADER_OPTIME) + last_lsn = last_lsn and last_lsn['Value'] + slots = None + + try: + last_lsn = int(last_lsn) + except Exception: + last_lsn = 0 # get list of members members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1] @@ -366,9 +381,9 @@ class Consul(AbstractDCS): sync = nodes.get(self._SYNC) sync = SyncState.from_node(sync and sync['ModifyIndex'], sync and sync['Value']) - return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) + return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots) except NotFound: - return Cluster(None, None, None, None, [], None, None, None) + return Cluster(None, None, None, None, [], None, None, None, None) except Exception: logger.exception('get_cluster') raise ConsulError('Consul is not responding properly') @@ -491,8 +506,12 @@ class Consul(AbstractDCS): return self._client.kv.put(self.config_path, value, cas=index) @catch_consul_errors - def _write_leader_optime(self, last_operation): - return self._client.kv.put(self.leader_optime_path, last_operation) + def _write_leader_optime(self, last_lsn): + return self._client.kv.put(self.leader_optime_path, last_lsn) + + @catch_consul_errors + def _write_status(self, value): + return self._client.kv.put(self.status_path, value) @catch_consul_errors def _update_leader(self): diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index b6e20ba0..5dfffad7 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -602,9 +602,24 @@ class Etcd(AbstractEtcd): history = nodes.get(self._HISTORY) history = history and TimelineHistory.from_node(history.modifiedIndex, history.value) - # get last leader operation - last_leader_operation = nodes.get(self._LEADER_OPTIME) - last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value) + # get last know leader lsn and slots + status = nodes.get(self._STATUS) + if status: + try: + status = json.loads(status.value) + last_lsn = status.get(self._OPTIME) + slots = status.get('slots') + except Exception: + slots = last_lsn = None + else: + last_lsn = nodes.get(self._LEADER_OPTIME) + last_lsn = last_lsn and last_lsn.value + slots = None + + try: + last_lsn = int(last_lsn) + except Exception: + last_lsn = 0 # get list of members members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1] @@ -626,9 +641,9 @@ class Etcd(AbstractEtcd): sync = nodes.get(self._SYNC) sync = SyncState.from_node(sync and sync.modifiedIndex, sync and sync.value) - cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) + cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots) except etcd.EtcdKeyNotFound: - cluster = Cluster(None, None, None, None, [], None, None, None) + cluster = Cluster(None, None, None, None, [], None, None, None, None) except Exception as e: self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly')) self._has_failed = False @@ -665,8 +680,12 @@ class Etcd(AbstractEtcd): return self._client.write(self.config_path, value, prevIndex=index or 0) @catch_etcd_errors - def _write_leader_optime(self, last_operation): - return self._client.set(self.leader_optime_path, last_operation) + def _write_leader_optime(self, last_lsn): + return self._client.set(self.leader_optime_path, last_lsn) + + @catch_etcd_errors + def _write_status(self, value): + return self._client.set(self.status_path, value) @catch_etcd_errors def _update_leader(self): diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 2d9c0656..51786676 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -654,9 +654,24 @@ class Etcd3(AbstractEtcd): history = nodes.get(self._HISTORY) history = history and TimelineHistory.from_node(history['mod_revision'], history['value']) - # get last leader operation - last_leader_operation = nodes.get(self._LEADER_OPTIME) - last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value']) + # get last know leader lsn and slots + status = nodes.get(self._STATUS) + if status: + try: + status = json.loads(status['value']) + last_lsn = status.get(self._OPTIME) + slots = status.get('slots') + except Exception: + slots = last_lsn = None + else: + last_lsn = nodes.get(self._LEADER_OPTIME) + last_lsn = last_lsn and last_lsn['value'] + slots = None + + try: + last_lsn = int(last_lsn) + except Exception: + last_lsn = 0 # get list of members members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1] @@ -680,7 +695,7 @@ class Etcd3(AbstractEtcd): sync = nodes.get(self._SYNC) sync = SyncState.from_node(sync and sync['mod_revision'], sync and sync['value']) - cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) + cluster = Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots) except UnsupportedEtcdVersion: raise except Exception as e: @@ -741,8 +756,12 @@ class Etcd3(AbstractEtcd): return self._client.put(self.config_path, value, mod_revision=index) @catch_etcd_errors - def _write_leader_optime(self, last_operation): - return self._client.put(self.leader_optime_path, last_operation) + def _write_leader_optime(self, last_lsn): + return self._client.put(self.leader_optime_path, last_lsn) + + @catch_etcd_errors + def _write_status(self, value): + return self._client.put(self.status_path, value) @catch_etcd_errors def _update_leader(self): diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index abd98a54..1a659036 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -724,9 +724,19 @@ class Kubernetes(AbstractDCS): self._leader_resource_version = metadata.resource_version if metadata else None annotations = metadata and metadata.annotations or {} - # get last leader operation - last_leader_operation = annotations.get(self._OPTIME) - last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation) + # get last known leader lsn + last_lsn = annotations.get(self._OPTIME) + try: + last_lsn = 0 if last_lsn is None else int(last_lsn) + except Exception: + last_lsn = 0 + + # get permanent slots state (confirmed_flush_lsn) + slots = annotations.get('slots') + try: + slots = slots and json.loads(slots) + except Exception: + slots = None # get leader leader_record = {n: annotations.get(n) for n in (self._LEADER, 'acquireTime', @@ -760,7 +770,7 @@ class Kubernetes(AbstractDCS): metadata = sync and sync.metadata sync = SyncState.from_node(metadata and metadata.resource_version, metadata and metadata.annotations) - return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) + return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots) except Exception: logger.exception('get_cluster') raise KubernetesError('Kubernetes API is not responding properly') @@ -881,7 +891,10 @@ class Kubernetes(AbstractDCS): return logger.exception('create_config_service failed') self._should_create_config_service = False - def _write_leader_optime(self, last_operation): + def _write_leader_optime(self, last_lsn): + """Unused""" + + def _write_status(self, value): """Unused""" def _update_leader(self): @@ -931,7 +944,7 @@ class Kubernetes(AbstractDCS): return self.patch_or_create(self.leader_path, annotations, kind_resource_version, ips=ips, retry=_retry) - def update_leader(self, last_operation, access_is_restricted=False): + def update_leader(self, last_lsn, slots=None): kind = self._kinds.get(self.leader_path) kind_annotations = kind and kind.metadata.annotations or {} @@ -943,12 +956,12 @@ class Kubernetes(AbstractDCS): annotations = {self._LEADER: self._name, 'ttl': str(self._ttl), 'renewTime': now, 'acquireTime': leader_observed_record.get('acquireTime') or now, 'transitions': leader_observed_record.get('transitions') or '0'} - if last_operation: - annotations[self._OPTIME] = last_operation + if last_lsn: + annotations[self._OPTIME] = str(last_lsn) + annotations['slots'] = json.dumps(slots) if slots else None resource_version = kind and kind.metadata.resource_version - ips = [] if access_is_restricted else self.__ips - return self._update_leader_with_retry(annotations, resource_version, ips) + return self._update_leader_with_retry(annotations, resource_version, self.__ips) def attempt_to_acquire_leader(self, permanent=False): now = datetime.datetime.now(tzutc).isoformat() @@ -1024,12 +1037,12 @@ class Kubernetes(AbstractDCS): def _delete_leader(self): """Unused""" - def delete_leader(self, last_operation=None): + def delete_leader(self, last_lsn=None): kind = self._kinds.get(self.leader_path) if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name: annotations = {self._LEADER: None} - if last_operation: - annotations[self._OPTIME] = last_operation + if last_lsn: + annotations[self._OPTIME] = last_lsn self.patch_or_create(self.leader_path, annotations, kind.metadata.resource_version, True, False, []) self.reset_cluster() diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index 811e3ff6..f94e6e64 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -4,11 +4,12 @@ import os import threading import time -from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory -from ..utils import validate_directory from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON from pysyncobj.transport import Node, TCPTransport, CONNECTION_STATE +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory +from ..utils import validate_directory + logger = logging.getLogger(__name__) @@ -328,7 +329,7 @@ class Raft(AbstractDCS): prefix = self.client_path('') response = self._sync_obj.get(prefix, recursive=True) if not response: - return Cluster(None, None, None, None, [], None, None, None) + return Cluster(None, None, None, None, [], None, None, None, None) nodes = {os.path.relpath(key, prefix).replace('\\', '/'): value for key, value in response.items()} # get initialize flag @@ -343,9 +344,24 @@ class Raft(AbstractDCS): history = nodes.get(self._HISTORY) history = history and TimelineHistory.from_node(history['index'], history['value']) - # get last leader operation - last_leader_operation = nodes.get(self._LEADER_OPTIME) - last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value']) + # get last know leader lsn and slots + status = nodes.get(self._STATUS) + if status: + try: + status = json.loads(status['value']) + last_lsn = status.get(self._OPTIME) + slots = status.get('slots') + except Exception: + slots = last_lsn = None + else: + last_lsn = nodes.get(self._LEADER_OPTIME) + last_lsn = last_lsn and last_lsn['value'] + slots = None + + try: + last_lsn = int(last_lsn) + except Exception: + last_lsn = 0 # get list of members members = [self.member(k, n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1] @@ -366,10 +382,13 @@ class Raft(AbstractDCS): sync = nodes.get(self._SYNC) sync = SyncState.from_node(sync and sync['index'], sync and sync['value']) - return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) + return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots) - def _write_leader_optime(self, last_operation): - return self._sync_obj.set(self.leader_optime_path, last_operation, timeout=1) + def _write_leader_optime(self, last_lsn): + return self._sync_obj.set(self.leader_optime_path, last_lsn, timeout=1) + + def _write_status(self, value): + return self._sync_obj.set(self.status_path, value, timeout=1) def _update_leader(self): ret = self._sync_obj.set(self.leader_path, self._name, ttl=self._ttl, prevValue=self._name) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index a7732cd4..3be8adb4 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -6,9 +6,10 @@ import time from kazoo.client import KazooClient, KazooState, KazooRetry from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.handlers.threading import SequentialThreadingHandler -from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory -from patroni.exceptions import DCSError -from patroni.utils import deep_compare + +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory +from ..exceptions import DCSError +from ..utils import deep_compare logger = logging.getLogger(__name__) @@ -74,7 +75,7 @@ class ZooKeeper(AbstractDCS): self._client.add_listener(self.session_listener) self._fetch_cluster = True - self._fetch_optime = True + self._fetch_status = True self._orig_kazoo_connect = self._client._connection._connect self._client._connection._connect = self._kazoo_connect @@ -100,13 +101,13 @@ class ZooKeeper(AbstractDCS): if state in [KazooState.SUSPENDED, KazooState.LOST]: self.cluster_watcher(None) - def optime_watcher(self, event): - self._fetch_optime = True + def status_watcher(self, event): + self._fetch_status = True self.event.set() def cluster_watcher(self, event): self._fetch_cluster = True - self.optime_watcher(event) + self.status_watcher(event) def reload_config(self, config): self.set_retry_timeout(config['retry_timeout']) @@ -151,11 +152,29 @@ class ZooKeeper(AbstractDCS): except NoNodeError: return None - def get_leader_optime(self, leader): - watch = self.optime_watcher if not leader or leader.name != self._name else None - optime = self.get_node(self.leader_optime_path, watch) - self._fetch_optime = False - return optime and int(optime[0]) or 0 + def get_status(self, leader): + watch = self.status_watcher if not leader or leader.name != self._name else None + + status = self.get_node(self.status_path, watch) + if status: + try: + status = json.loads(status[0]) + last_lsn = status.get(self._OPTIME) + slots = status.get('slots') + except Exception: + slots = last_lsn = None + else: + last_lsn = self.get_node(self.leader_optime_path, watch) + last_lsn = last_lsn and last_lsn[0] + slots = None + + try: + last_lsn = int(last_lsn) + except Exception: + last_lsn = 0 + + self._fetch_status = False + return last_lsn, slots @staticmethod def member(name, value, znode): @@ -218,14 +237,14 @@ class ZooKeeper(AbstractDCS): leader = Leader(leader[1].version, leader[1].ephemeralOwner, member) self._fetch_cluster = member.index == -1 - # get last leader operation - last_leader_operation = self._OPTIME in nodes and self.get_leader_optime(leader) + # get last known leader lsn and slots + last_lsn, slots = self.get_status(leader) # failover key failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None failover = failover and Failover.from_node(failover[1].version, failover[0]) - return Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history) + return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots) def _load_cluster(self): cluster = self.cluster @@ -237,12 +256,13 @@ class ZooKeeper(AbstractDCS): self.cluster_watcher(None) raise ZooKeeperError('ZooKeeper in not responding properly') # Optime ZNode was updated or doesn't exist and we are not leader - elif (self._fetch_optime and not self._fetch_cluster or not cluster.last_leader_operation) and\ + elif (self._fetch_status and not self._fetch_cluster or not cluster.last_lsn + or cluster.has_permanent_logical_slots(self._name, False) and not cluster.slots) and\ not (cluster.leader and cluster.leader.name == self._name): try: - optime = self.get_leader_optime(cluster.leader) - cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, optime, - cluster.members, cluster.failover, cluster.sync, cluster.history) + last_lsn, slots = self.get_status(cluster.leader) + cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, last_lsn, + cluster.members, cluster.failover, cluster.sync, cluster.history, slots) except Exception: pass return cluster @@ -336,8 +356,11 @@ class ZooKeeper(AbstractDCS): def take_leader(self): return self.attempt_to_acquire_leader() - def _write_leader_optime(self, last_operation): - return self._set_or_create(self.leader_optime_path, last_operation) + def _write_leader_optime(self, last_lsn): + return self._set_or_create(self.leader_optime_path, last_lsn) + + def _write_status(self, value): + return self._set_or_create(self.status_path, value) def _update_leader(self): return True @@ -374,6 +397,6 @@ class ZooKeeper(AbstractDCS): def watch(self, leader_index, timeout): ret = super(ZooKeeper, self).watch(leader_index, timeout) - if ret and not self._fetch_optime: + if ret and not self._fetch_status: self._fetch_cluster = True return ret or self._fetch_cluster diff --git a/patroni/ha.py b/patroni/ha.py index 6adf437c..8d94a94d 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -66,7 +66,6 @@ class Ha(object): self.old_cluster = None self._is_leader = False self._is_leader_lock = RLock() - self._leader_access_is_restricted = False self._was_paused = False self._leader_timeline = None self.recovering = False @@ -121,16 +120,12 @@ class Ha(object): def is_leader(self): with self._is_leader_lock: - return self._is_leader > time.time() and not self._leader_access_is_restricted + return self._is_leader > time.time() def set_is_leader(self, value): with self._is_leader_lock: self._is_leader = time.time() + self.dcs.ttl if value else 0 - def set_leader_access_is_restricted(self, value): - with self._is_leader_lock: - self._leader_access_is_restricted = value - def load_cluster_from_dcs(self): cluster = self.dcs.get_cluster() @@ -145,20 +140,20 @@ class Ha(object): self._leader_timeline = None if cluster.is_unlocked() else cluster.leader.timeline def acquire_lock(self): - self.set_leader_access_is_restricted(self.cluster.has_permanent_logical_slots(self.state_handler.name)) ret = self.dcs.attempt_to_acquire_leader() self.set_is_leader(ret) return ret def update_lock(self, write_leader_optime=False): - last_operation = None + last_lsn = slots = None if write_leader_optime: try: - last_operation = self.state_handler.last_operation() + last_lsn = self.state_handler.last_operation() + slots = self.state_handler.slots() except Exception: logger.exception('Exception when called state_handler.last_operation()') try: - ret = self.dcs.update_leader(last_operation, self._leader_access_is_restricted) + ret = self.dcs.update_leader(last_lsn, slots) except Exception: logger.exception('Unexpected exception raised from update_leader, please report it as a BUG') ret = False @@ -613,8 +608,6 @@ class Ha(object): return 'Postponing promotion because synchronous replication state was updated by somebody else' self.state_handler.config.set_synchronous_standby(['*'] if self.is_synchronous_mode_strict() else []) if self.state_handler.role != 'master': - self.set_leader_access_is_restricted(self.cluster.has_permanent_logical_slots(self.state_handler.name)) - def on_success(): self._rewind.reset_state() logger.info("cleared rewind state after becoming the leader") @@ -622,8 +615,7 @@ class Ha(object): with self._async_response: self._async_response.reset() self._async_executor.try_run_async('promote', self.state_handler.promote, - args=(self.dcs.loop_wait, self._async_response, on_success, - self._leader_access_is_restricted)) + args=(self.dcs.loop_wait, self._async_response, on_success)) return promote_message def fetch_node_status(self, member): @@ -653,14 +645,13 @@ class Ha(object): :param wal_position: Current wal position. :returns True when node is lagging """ - lag = (self.cluster.last_leader_operation or 0) - wal_position + lag = (self.cluster.last_lsn or 0) - wal_position return lag > self.patroni.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.""" - # We don't call `last_operation()` here because it returns a string - _, my_wal_position, _ = self.state_handler.timeline_wal_position() + my_wal_position = self.state_handler.last_operation() if check_replication_lag and self.is_lagging(my_wal_position): logger.info('My wal position exceeds maximum replication lag') return False # Too far behind last reported wal position on master @@ -802,13 +793,13 @@ class Ha(object): return self._is_healthiest_node(members.values()) - def _delete_leader(self, last_operation=None): + def _delete_leader(self, last_lsn=None): self.set_is_leader(False) - self.dcs.delete_leader(last_operation) + self.dcs.delete_leader(last_lsn) self.dcs.reset_cluster() - def release_leader_key_voluntarily(self, last_operation=None): - self._delete_leader(last_operation) + def release_leader_key_voluntarily(self, last_lsn=None): + self._delete_leader(last_lsn) self.touch_member() logger.info("Leader key released") @@ -989,11 +980,6 @@ class Ha(object): self._delete_leader() return 'removed leader lock because postgres is not running as master' - if self.state_handler.is_leader() and self._leader_access_is_restricted: - self.state_handler.slots_handler.sync_replication_slots(self.cluster) - self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE) - self.set_leader_access_is_restricted(False) - if self.update_lock(True): msg = self.process_manual_failover_from_leader() if msg is not None: @@ -1261,7 +1247,6 @@ class Ha(object): self.cancel_initialization() self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid) self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':'))) - self.state_handler.slots_handler.sync_replication_slots(self.cluster) self.dcs.take_leader() self.set_is_leader(True) self.state_handler.call_nowait(ACTION_ON_START) @@ -1317,8 +1302,8 @@ class Ha(object): def _run_cycle(self): dcs_failed = False try: - self.state_handler.reset_cluster_info_state() self.load_cluster_from_dcs() + self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover) if self.is_paused(): self.watchdog.disable() @@ -1424,20 +1409,28 @@ class Ha(object): try: if self.cluster.is_unlocked(): - return self.process_unhealthy_cluster() + ret = self.process_unhealthy_cluster() else: msg = self.process_healthy_cluster() - return self.evaluate_scheduled_restart() or msg + ret = self.evaluate_scheduled_restart() or msg finally: # we might not have a valid PostgreSQL connection here if another thread # stops PostgreSQL, therefore, we only reload replication slots if no # asynchronous processes are running (should be always the case for the master) if not self._async_executor.busy and not self.state_handler.is_starting(): - self.state_handler.slots_handler.sync_replication_slots(self.cluster) + create_slots = self.state_handler.slots_handler.sync_replication_slots(self.cluster, + self.patroni.nofailover) if not self.state_handler.cb_called: if not self.state_handler.is_leader(): self._rewind.trigger_check_diverged_lsn() self.state_handler.call_nowait(ACTION_ON_START) + if create_slots and self.cluster.leader: + err = self._async_executor.try_run_async('copy_logical_slots', + self.state_handler.slots_handler.copy_logical_slots, + args=(self.cluster.leader, create_slots)) + if not err: + ret = 'Copying logical slots {0} from the primary'.format(create_slots) + return ret except DCSError: dcs_failed = True logger.error('Error communicating with DCS') @@ -1468,7 +1461,7 @@ class Ha(object): self.watchdog.disable() elif not self._join_aborted: # FIXME: If stop doesn't reach safepoint quickly enough keepalive is triggered. If shutdown checkpoint - # takes longer than ttl, then leader key is lost and replication might not have sent out all xlog. + # takes longer than ttl, then leader key is lost and replication might not have sent out all WAL. # This might not be the desired behavior of users, as a graceful shutdown of the host can mean lost data. # We probably need to something smarter here. disable_wd = self.watchdog.disable if self.watchdog.is_running else None diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index aced2d7c..b33fff8b 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -10,19 +10,20 @@ from contextlib import contextmanager from copy import deepcopy from dateutil import tz from datetime import datetime -from patroni.postgresql.callback_executor import CallbackExecutor -from patroni.postgresql.bootstrap import Bootstrap -from patroni.postgresql.cancellable import CancellableSubprocess -from patroni.postgresql.config import ConfigHandler, mtime -from patroni.postgresql.connection import Connection, get_connection_cursor -from patroni.postgresql.misc import parse_history, parse_lsn, postgres_major_version_to_int -from patroni.postgresql.postmaster import PostmasterProcess -from patroni.postgresql.slots import SlotsHandler -from patroni.exceptions import PostgresConnectionException -from patroni.utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int from psutil import TimeoutExpired from threading import current_thread, Lock +from .callback_executor import CallbackExecutor +from .bootstrap import Bootstrap +from .cancellable import CancellableSubprocess +from .config import ConfigHandler, mtime +from .connection import Connection, get_connection_cursor +from .misc import parse_history, parse_lsn, postgres_major_version_to_int +from .postmaster import PostmasterProcess +from .slots import SlotsHandler +from ..exceptions import PostgresConnectionException +from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int + logger = logging.getLogger(__name__) @@ -54,7 +55,7 @@ class Postgresql(object): "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, " # master timeline "CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 " "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint END, " # write_lsn - "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint, " + "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint, " "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, " "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()") @@ -101,6 +102,8 @@ class Postgresql(object): self._state_entry_timestamp = None self._cluster_info_state = {} + self._has_permanent_logical_slots = True + self._enforce_hot_standby_feedback = False self._cached_replica_timeline = None # Last known running process @@ -152,14 +155,18 @@ class Postgresql(object): @property def cluster_info_query(self): if self._major_version >= 90600: + extra = "(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, " +\ + "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" + \ + " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"\ + if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL" extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END," - " slot_name, conninfo FROM pg_catalog.pg_stat_get_wal_receiver()") + " slot_name, conninfo, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra) if self.role == 'standby_leader': extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()" else: extra = "0" + extra else: - extra = "0, NULL, NULL, NULL" + extra = "0, NULL, NULL, NULL, NULL" return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra) @@ -299,16 +306,36 @@ class Postgresql(object): replica_methods = self.create_replica_methods return any(self.replica_method_can_work_without_replication_connection(m) for m in replica_methods) - def reset_cluster_info_state(self): + @property + def enforce_hot_standby_feedback(self): + return self._enforce_hot_standby_feedback + + def set_enforce_hot_standby_feedback(self, value): + # If we enable or disable the hot_standby_feedback we need to update postgresql.conf and reload + if self._enforce_hot_standby_feedback != value: + self._enforce_hot_standby_feedback = value + if self.is_running(): + self.config.write_postgresql_conf() + self.reload() + + def reset_cluster_info_state(self, cluster, nofailover=None): self._cluster_info_state = {} + if cluster and cluster.config and cluster.config.modify_index: + self._has_permanent_logical_slots =\ + cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version) + self.set_enforce_hot_standby_feedback( + self._has_permanent_logical_slots or + cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) def _cluster_info_state_get(self, name): if not self._cluster_info_state: try: result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone() - self._cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location', - 'received_location', 'replay_paused', 'pg_control_timeline', - 'received_tli', 'slot_name', 'conninfo'], result)) + cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location', + 'received_location', 'replay_paused', 'pg_control_timeline', + 'received_tli', 'slot_name', 'conninfo', 'slots'], result)) + cluster_info_state['slots'] = self.slots_handler.process_permanent_slots(cluster_info_state['slots']) + self._cluster_info_state = cluster_info_state except RetryFailedError as e: # SELECT failed two times self._cluster_info_state = {'error': str(e)} if not self.is_starting() and self.pg_isready() == STATE_REJECT: @@ -325,6 +352,9 @@ class Postgresql(object): def received_location(self): return self._cluster_info_state_get('received_location') + def slots(self): + return self._cluster_info_state_get('slots') + def primary_slot_name(self): return self._cluster_info_state_get('slot_name') @@ -362,7 +392,7 @@ class Postgresql(object): return self._postmaster_proc self._postmaster_proc = None - # we noticed that postgres was restarted, force syncing of replication + # we noticed that postgres was restarted, force syncing of replication slots and check of logical slots self.slots_handler.schedule() self._postmaster_proc = PostmasterProcess.from_pidfile(self._data_dir) @@ -812,7 +842,7 @@ class Postgresql(object): logger.info('pre_promote script `%s` exited with %s', cmd, ret) return ret == 0 - def promote(self, wait_seconds, task, on_success=None, access_is_restricted=False): + def promote(self, wait_seconds, task, on_success=None): if self.role == 'master': return True @@ -829,13 +859,14 @@ class Postgresql(object): logger.info("PostgreSQL promote cancelled.") return False + self.slots_handler.on_promote() + ret = self.pg_ctl('promote', '-W') if ret: self.set_role('master') if on_success is not None: on_success() - if not access_is_restricted: - self.call_nowait(ACTION_ON_ROLE_CHANGE) + self.call_nowait(ACTION_ON_ROLE_CHANGE) ret = self._wait_promote(wait_seconds) return ret @@ -873,8 +904,8 @@ class Postgresql(object): return None def last_operation(self): - return str(self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position'), - self.received_location(), self.replayed_location())) + return self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position'), + self.received_location(), self.replayed_location()) def configure_server_parameters(self): self._major_version = self.get_major_version() diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index d6e70bc7..7ad794ef 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -387,15 +387,21 @@ class ConfigHandler(object): if 'custom_conf' not in self._config and not os.path.exists(self._postgresql_base_conf): os.rename(self._postgresql_conf, self._postgresql_base_conf) + configuration = configuration or self._server_parameters.copy() + # In case we are using custom bootstrap from spilo image with PITR it fails if it contains increasing # values like Max_connections. We disable hot_standby so it will accept increasing values. if self._postgresql.bootstrap.running_custom_bootstrap: configuration['hot_standby'] = 'off' + # Due to the permanent logical replication slots configured we have to enable hot_standby_feedback + if self._postgresql.enforce_hot_standby_feedback: + configuration['hot_standby_feedback'] = 'on' + with ConfigWriter(self._postgresql_conf) as f: include = self._config.get('custom_conf') or self._postgresql_base_conf_name f.writeline("include '{0}'\n".format(ConfigWriter.escape(include))) - for name, value in sorted((configuration or self._server_parameters).items()): + for name, value in sorted((configuration).items()): value = transform_postgresql_parameter_value(self._postgresql.major_version, name, value) if (not self._postgresql.bootstrap.running_custom_bootstrap or name != 'hba_file') \ and name not in self._RECOVERY_PARAMETERS and value is not None: diff --git a/patroni/postgresql/misc.py b/patroni/postgresql/misc.py index f5caffbd..1d93d3de 100644 --- a/patroni/postgresql/misc.py +++ b/patroni/postgresql/misc.py @@ -68,3 +68,8 @@ def parse_history(data): yield values except (IndexError, ValueError): logger.exception('Exception when parsing timeline history line "%s"', values) + + +def format_lsn(lsn, full=False): + template = '{0:X}/{1:08X}' if full else '{0:X}/{1:X}' + return template.format(lsn >> 32, lsn & 0xFFFFFFFF) diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 82591b51..519b0b24 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -7,7 +7,7 @@ import subprocess from threading import Lock, Thread from .connection import get_connection_cursor -from .misc import parse_history, parse_lsn +from .misc import format_lsn, parse_history, parse_lsn from ..async_executor import CriticalTask from ..dcs import Leader @@ -17,11 +17,6 @@ REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECKPOINT': 1, 'CHECK': 2, 'NE 'NOT_NEED': 4, 'SUCCESS': 5, 'FAILED': 6}) -def format_lsn(lsn, full=False): - template = '{0:X}/{1:08X}' if full else '{0:X}/{1:X}' - return template.format(lsn >> 32, lsn & 0xFFFFFFFF) - - class Rewind(object): def __init__(self, postgresql): diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 3738a25d..cee319d9 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -1,14 +1,33 @@ +import errno import logging +import os +import shutil -from patroni.postgresql.connection import get_connection_cursor from collections import defaultdict +from contextlib import contextmanager + +from .connection import get_connection_cursor +from .misc import format_lsn logger = logging.getLogger(__name__) -def compare_slots(s1, s2): +def compare_slots(s1, s2, dbid='database'): return s1['type'] == s2['type'] and (s1['type'] == 'physical' or - s1['database'] == s2['database'] and s1['plugin'] == s2['plugin']) + s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin']) + + +def fsync_dir(path): + if os.name != 'nt': + fd = os.open(path, os.O_DIRECTORY) + try: + os.fsync(fd) + except OSError as e: + # Some filesystems don't like fsyncing directories and raise EINVAL. Ignoring it is usually safe. + if e.errno != errno.EINVAL: + raise + finally: + os.close(fd) class SlotsHandler(object): @@ -16,22 +35,66 @@ class SlotsHandler(object): def __init__(self, postgresql): self._postgresql = postgresql self._replication_slots = {} # already existing replication slots + self._unready_logical_slots = set() self.schedule() def _query(self, sql, *params): return self._postgresql.query(sql, *params, retry=False) + @staticmethod + def _copy_items(src, dst, keys=None): + dst.update({key: src[key] for key in keys or ('datoid', 'catalog_xmin', 'confirmed_flush_lsn')}) + + def process_permanent_slots(self, slots): + """This methods solves three problems at once (I know, it is weird). + + The cluster_info_query from `Postgresql` is executed every HA loop and returns + information about all replication slots that exists on the current host. + Based on this information we perform the following actions: + 1. For the primary we want to expose to DCS permanent logical slots, therefore the method + builds (and returns) a dict, that maps permanent logical slot names and confirmed_flush_lsns. + 2. This method also detects if one of the previously known permanent slots got missing and schedules resync. + 3. Updates the local cache with the fresh catalog_xmin and confirmed_flush_lsn for every known slot. + This info is used when performing the check of logical slot readiness on standbys. + """ + ret = {} + + slots = {slot['slot_name']: slot for slot in slots or []} + if slots: + for name, value in slots.items(): + if name in self._replication_slots: + if compare_slots(value, self._replication_slots[name], 'datoid'): + if value['type'] == 'logical': + ret[name] = value['confirmed_flush_lsn'] + self._copy_items(value, self._replication_slots[name]) + else: + self._schedule_load_slots = True + + # It could happen that the slots was deleted in the background, we want to detect this case + if any(name not in slots for name in self._replication_slots.keys()): + self._schedule_load_slots = True + + return ret + def load_replication_slots(self): if self._postgresql.major_version >= 90400 and self._schedule_load_slots: replication_slots = {} - cursor = self._query('SELECT slot_name, slot_type, plugin, database FROM pg_catalog.pg_replication_slots') + extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"\ + if self._postgresql.major_version >= 100000 else "" + cursor = self._query('SELECT slot_name, slot_type, plugin, database, datoid' + '{0} FROM pg_catalog.pg_replication_slots'.format(extra)) for r in cursor: value = {'type': r[1]} if r[1] == 'logical': - value.update({'plugin': r[2], 'database': r[3]}) + value.update(plugin=r[2], database=r[3], datoid=r[4]) + if self._postgresql.major_version >= 100000: + value.update(catalog_xmin=r[5], confirmed_flush_lsn=r[6]) replication_slots[r[0]] = value self._replication_slots = replication_slots self._schedule_load_slots = False + if self._force_readiness_check: + self._unready_logical_slots = set(n for n, v in replication_slots.items() if v['type'] == 'logical') + self._force_readiness_check = False def ignore_replication_slot(self, cluster, name): slot = self._replication_slots[name] @@ -47,65 +110,188 @@ class SlotsHandler(object): # In normal situation rowcount should be 1, otherwise either slot doesn't exists or it is still active return cursor.rowcount == 1 - def sync_replication_slots(self, cluster): + def _drop_incorrect_slots(self, cluster, slots): + # drop old replication slots which are not presented in desired slots + for name in set(self._replication_slots) - set(slots): + if not self.ignore_replication_slot(cluster, name) and not self.drop_replication_slot(name): + logger.error("Failed to drop replication slot '%s'", name) + self._schedule_load_slots = True + + for name, value in slots.items(): + if name in self._replication_slots and not compare_slots(value, self._replication_slots[name]): + logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s", + name, self._replication_slots[name], value) + if self.drop_replication_slot(name): + self._replication_slots.pop(name) + else: + logger.error("Failed to drop replication slot '%s'", name) + self._schedule_load_slots = True + + def _ensure_physical_slots(self, slots): + immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else '' + for name, value in slots.items(): + if name not in self._replication_slots and value['type'] == 'physical': + try: + self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" + + " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" + + " WHERE slot_type = 'physical' AND slot_name = %s)").format( + immediately_reserve), name, name) + except Exception: + logger.exception("Failed to create physical replication slot '%s'", name) + self._schedule_load_slots = True + + @contextmanager + def _get_local_connection_cursor(self, database): + conn_kwargs = self._postgresql.config.local_connect_kwargs + conn_kwargs['database'] = database + with get_connection_cursor(**conn_kwargs) as cur: + yield cur + + def _ensure_logical_slots_primary(self, slots): + # Group logical slots to be created by database name + logical_slots = defaultdict(dict) + for name, value in slots.items(): + if value['type'] == 'logical': + # If the logical already exists, copy some information about it into the original structure + if self._replication_slots.get(name, {}).get('datoid'): + self._copy_items(self._replication_slots[name], value) + else: + logical_slots[value['database']][name] = value + + # Create new logical slots + for database, values in logical_slots.items(): + with self._get_local_connection_cursor(database) as cur: + for name, value in values.items(): + try: + cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" + + " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" + + " WHERE slot_type = 'logical' AND slot_name = %s)", + (name, value['plugin'], name)) + except Exception as e: + logger.error("Failed to create logical replication slot '%s' plugin='%s': %r", + name, value['plugin'], e) + slots.pop(name) + self._schedule_load_slots = True + + def _ensure_logical_slots_replica(self, cluster, slots): + advance_slots = defaultdict(dict) # Group logical slots to be advanced by database name + create_slots = [] # And collect logical slots to be created on the replica + for name, value in slots.items(): + if value['type'] == 'logical': + # If the logical already exists, copy some information about it into the original structure + if self._replication_slots.get(name, {}).get('datoid'): + self._copy_items(self._replication_slots[name], value) + if name in cluster.slots: + try: # Skip slots that doesn't need to be advanced + if value['confirmed_flush_lsn'] < int(cluster.slots[name]): + advance_slots[value['database']][name] = value + except Exception as e: + logger.error('Failed to parse "%s": %r', cluster.slots[name], e) + elif name in cluster.slots: # We want to copy only slots with feedback in a DCS + create_slots.append(name) + + # Advance logical slots + for database, values in advance_slots.items(): + with self._get_local_connection_cursor(database) as cur: + for name, value in values.items(): + try: + cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", + (name, format_lsn(int(cluster.slots[name])))) + except Exception as e: + logger.exception("Failed to advance logical replication slot '%s': %r", name, e) + self._schedule_load_slots = True + return create_slots + + def sync_replication_slots(self, cluster, nofailover, replicatefrom=None): + ret = None if self._postgresql.major_version >= 90400 and cluster.config: try: self.load_replication_slots() - slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role) + slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role, + nofailover, self._postgresql.major_version, True) - # drop old replication slots which are not presented in desired slots - for name in set(self._replication_slots) - set(slots): - if not self.ignore_replication_slot(cluster, name) and not self.drop_replication_slot(name): - logger.error("Failed to drop replication slot '%s'", name) - self._schedule_load_slots = True + self._drop_incorrect_slots(cluster, slots) - immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else '' + self._ensure_physical_slots(slots) - logical_slots = defaultdict(dict) - for name, value in slots.items(): - if name in self._replication_slots and not compare_slots(value, self._replication_slots[name]): - logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s", - name, self._replication_slots[name], value) - if not self.drop_replication_slot(name): - logger.error("Failed to drop replication slot '%s'", name) - self._schedule_load_slots = True - continue - self._replication_slots.pop(name) - if name not in self._replication_slots: - if value['type'] == 'physical': - try: - self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" + - " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" + - " WHERE slot_type = 'physical' AND slot_name = %s)").format( - immediately_reserve), name, name) - except Exception: - logger.exception("Failed to create physical replication slot '%s'", name) - self._schedule_load_slots = True - elif value['type'] == 'logical' and name not in self._replication_slots: - logical_slots[value['database']][name] = value + if self._postgresql.is_leader(): + self._unready_logical_slots.clear() + self._ensure_logical_slots_primary(slots) + elif cluster.slots and slots: + self.check_logical_slots_readiness(cluster, nofailover, replicatefrom) + + ret = self._ensure_logical_slots_replica(cluster, slots) - # create new logical slots - for database, values in logical_slots.items(): - conn_kwargs = self._postgresql.config.local_connect_kwargs - conn_kwargs['database'] = database - with get_connection_cursor(**conn_kwargs) as cur: - for name, value in values.items(): - try: - cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" + - " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" + - " WHERE slot_type = 'logical' AND slot_name = %s)", - (name, value['plugin'], name)) - except Exception: - logger.exception("Failed to create logical replication slot '%s' plugin='%s'", - name, value['plugin']) - self._schedule_load_slots = True self._replication_slots = slots except Exception: logger.exception('Exception when changing replication slots') self._schedule_load_slots = True + return ret + + @contextmanager + def _get_leader_connection_cursor(self, leader): + conn_kwargs = leader.conn_kwargs(self._postgresql.config.rewind_credentials) + conn_kwargs['database'] = self._postgresql.database + with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur: + yield cur + + def check_logical_slots_readiness(self, cluster, nofailover, replicatefrom): + if self._unready_logical_slots: + slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom) + try: + with self._get_leader_connection_cursor(cluster.leader) as cur: + cur.execute("SELECT catalog_xmin FROM pg_catalog.pg_get_replication_slots()" + " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = %s", (slot_name,)) + if cur.rowcount < 1: + return logger.warning('Physical slot %s does not exist on the primary', slot_name) + catalog_xmin = cur.fetchone()[0] + except Exception as e: + return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e) + for name in list(self._unready_logical_slots): + value = self._replication_slots.get(name) + if not value or catalog_xmin <= value['catalog_xmin']: + self._unready_logical_slots.remove(name) + if value: + logger.info('Logical slot %s is safe to be used after a failover', name) + + def copy_logical_slots(self, leader, slots): + with self._get_leader_connection_cursor(leader) as cur: + try: + cur.execute("SELECT slot_name, catalog_xmin, " + "pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint, " + "pg_catalog.pg_read_binary_file('pg_replslot/' || slot_name || '/state')" + " FROM pg_catalog.pg_get_replication_slots() WHERE NOT pg_catalog.pg_is_in_recovery()" + " AND slot_name = ANY(%s)", (slots,)) + slots = {r[0]: {'catalog_xmin': r[1], 'confirmed_flush_lsn': r[2], 'data': r[3]} for r in cur} + except Exception as e: + logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e) + + if isinstance(slots, dict) and self._postgresql.stop(): + pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot') + for name, value in slots.items(): + slot_dir = os.path.join(pg_replslot_dir, name) + slot_tmp_dir = slot_dir + '.tmp' + if os.path.exists(slot_tmp_dir): + shutil.rmtree(slot_tmp_dir) + os.makedirs(slot_tmp_dir) + fsync_dir(slot_tmp_dir) + with open(os.path.join(slot_tmp_dir, 'state'), 'wb') as f: + f.write(value['data']) + f.flush() + os.fsync(f.fileno()) + os.rename(slot_tmp_dir, slot_dir) + fsync_dir(slot_dir) + self._unready_logical_slots.add(name) + fsync_dir(pg_replslot_dir) + self._postgresql.start() def schedule(self, value=None): if value is None: value = self._postgresql.major_version >= 90400 - self._schedule_load_slots = value + self._schedule_load_slots = self._force_readiness_check = value + + def on_promote(self): + if self._unready_logical_slots: + logger.warning('Logical replication slots that might be unsafe to use after promote: %s', + self._unready_logical_slots) diff --git a/patroni/utils.py b/patroni/utils.py index 0bbddb4d..1cfad919 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -405,7 +405,7 @@ def is_standby_cluster(config): def cluster_as_json(cluster): leader_name = cluster.leader.name if cluster.leader else None - xlog_location_cluster = cluster.last_leader_operation or 0 + cluster_lsn = cluster.last_lsn or 0 ret = {'members': []} for m in cluster.members: @@ -427,11 +427,11 @@ def cluster_as_json(cluster): member.update({n: m.data[n] for n in optional_attributes if n in m.data}) if m.name != leader_name: - xlog_location = m.data.get('xlog_location') - if xlog_location is None: + lsn = m.data.get('xlog_location') + if lsn is None: member['lag'] = 'unknown' - elif xlog_location_cluster >= xlog_location: - member['lag'] = xlog_location_cluster - xlog_location + elif cluster_lsn >= lsn: + member['lag'] = cluster_lsn - lsn else: member['lag'] = 0 diff --git a/tests/__init__.py b/tests/__init__.py index 23861881..4c433862 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -90,10 +90,14 @@ class MockCursor(object): raise psycopg2.OperationalError() elif sql.startswith('RetryFailedError'): raise RetryFailedError('retry') + elif sql.startswith('SELECT catalog_xmin'): + self.results = [(100, 501)] + elif sql.startswith('SELECT slot_name, catalog_xmin'): + self.results = [('ls', 100, 500, b'123456')] elif sql.startswith('SELECT slot_name'): - self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b')] + self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)] elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'): - self.results = [(1, 2, 1, 0, False, 1, 1, None, None)] + self.results = [(1, 2, 1, 0, False, 1, 1, None, None, [{"slot_name": "ls", "confirmed_flush_lsn": 12345}])] elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'): self.results = [(False, 2)] elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'): diff --git a/tests/test_api.py b/tests/test_api.py index ec91267e..079753f4 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -162,7 +162,7 @@ class TestRestApiHandler(unittest.TestCase): _authorization = '\nAuthorization: Basic dGVzdDp0ZXN0' def test_do_GET(self): - MockPatroni.dcs.cluster.last_leader_operation = 20 + MockPatroni.dcs.cluster.last_lsn = 20 MockRestApiServer(RestApiHandler, 'GET /replica') MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M') MockRestApiServer(RestApiHandler, 'GET /replica?lag=10MB') diff --git a/tests/test_consul.py b/tests/test_consul.py index a38a2869..6b5f773b 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -15,8 +15,7 @@ def kv_get(self, key, **kwargs): return None, None if key == 'service/good/leader': return '1', None - if key == 'service/good/': - return ('6429', + good_cls = ('6429', [{'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'failover', 'LockIndex': 0, 'ModifyIndex': 1334, 'Value': b''}, {'CreateIndex': 1334, 'Flags': 0, 'Key': key + 'initialize', 'LockIndex': 0, @@ -34,7 +33,17 @@ def kv_get(self, key, **kwargs): {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'optime/leader', 'LockIndex': 0, 'ModifyIndex': 6429, 'Value': b'4496294792'}, {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'sync', 'LockIndex': 0, - 'ModifyIndex': 6429, 'Value': b'{"leader": "leader", "sync_standby": null}'}]) + 'ModifyIndex': 6429, 'Value': b'{"leader": "leader", "sync_standby": null}'}, + {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'status', 'LockIndex': 0, + 'ModifyIndex': 6429, 'Value': b'{"optime":4496294792, "slots":{"ls":12345}}'}]) + if key == 'service/good/': + return good_cls + if key == 'service/broken/': + good_cls[1][-1]['Value'] = b'{' + return good_cls + if key == 'service/legacy/': + good_cls[1].pop() + return good_cls raise ConsulException @@ -109,6 +118,10 @@ class TestConsul(unittest.TestCase): self.assertIsInstance(self.c.get_cluster(), Cluster) self.c._base_path = '/service/fail' self.assertRaises(ConsulError, self.c.get_cluster) + self.c._base_path = '/service/broken' + self.assertIsInstance(self.c.get_cluster(), Cluster) + self.c._base_path = '/service/legacy' + self.assertIsInstance(self.c.get_cluster(), Cluster) self.c._base_path = '/service/good' self.c._session = 'fd4f44fe-2cac-bba5-a60b-304b51ff39b8' self.assertIsInstance(self.c.get_cluster(), Cluster) @@ -146,7 +159,7 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.Session, 'renew', Mock()) def test_update_leader(self): - self.c.update_leader(None) + self.c.update_leader(12345) @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True)) def test_delete_leader(self): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 9648d4ec..b96082ae 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -66,7 +66,13 @@ def etcd_read(self, key, **kwargs): "?application_name=http://127.0.0.1:8008/patroni", "expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30, "modifiedIndex": 20730, "createdIndex": 20730}], - "modifiedIndex": 1581, "createdIndex": 1581}], "modifiedIndex": 1581, "createdIndex": 1581}} + "modifiedIndex": 1581, "createdIndex": 1581}, + {"key": "/service/batman5/status", "value": '{"optime":2164261704,"slots":{"ls":12345}}', + "modifiedIndex": 1582, "createdIndex": 1582}], "modifiedIndex": 1581, "createdIndex": 1581}} + if key == '/service/legacy/': + response['node']['nodes'].pop() + if key == '/service/broken/': + response['node']['nodes'][-1]['value'] = '{' result = etcd.EtcdResult(**response) result.etcd_index = 0 return result @@ -246,6 +252,10 @@ class TestEtcd(unittest.TestCase): cluster = self.etcd.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertFalse(cluster.is_synchronous_mode()) + self.etcd._base_path = '/service/legacy' + self.assertIsInstance(self.etcd.get_cluster(), Cluster) + self.etcd._base_path = '/service/broken' + self.assertIsInstance(self.etcd.get_cluster(), Cluster) self.etcd._base_path = '/service/nocluster' cluster = self.etcd.get_cluster() self.assertIsInstance(cluster, Cluster) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 70f19e1d..abc593af 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -33,6 +33,8 @@ def mock_urlopen(self, method, url, **kwargs): "value": base64_encode('foo'), "lease": "bla", "mod_revision": '1'}, {"key": base64_encode('/patroni/test/members/foo'), "value": base64_encode('{}'), "lease": "123", "mod_revision": '1'}, + {"key": base64_encode('/patroni/test/members/bar'), + "value": base64_encode('{"version":"1.6.5"}'), "lease": "123", "mod_revision": '1'}, {"key": base64_encode('/patroni/test/failover'), "value": base64_encode('{}'), "mod_revision": '1'} ] }) @@ -172,6 +174,22 @@ class TestEtcd3(BaseTestEtcd3): self.assertIsInstance(self.etcd3.get_cluster(), Cluster) self.client._kv_cache = None with patch.object(urllib3.PoolManager, 'urlopen') as mock_urlopen: + mock_urlopen.return_value = MockResponse() + mock_urlopen.return_value.content = json.dumps({ + "header": {"revision": "1"}, + "kvs": [ + {"key": base64_encode('/patroni/test/status'), + "value": base64_encode('{"optime":1234567,"slots":{"ls":12345}}'), "mod_revision": '1'} + ] + }) + self.assertIsInstance(self.etcd3.get_cluster(), Cluster) + mock_urlopen.return_value.content = json.dumps({ + "header": {"revision": "1"}, + "kvs": [ + {"key": base64_encode('/patroni/test/status'), "value": base64_encode('{'), "mod_revision": '1'} + ] + }) + self.assertIsInstance(self.etcd3.get_cluster(), Cluster) mock_urlopen.side_effect = UnsupportedEtcdVersion('') self.assertRaises(UnsupportedEtcdVersion, self.etcd3.get_cluster) mock_urlopen.side_effect = SleepException() diff --git a/tests/test_ha.py b/tests/test_ha.py index b81b7bba..ed534b27 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -38,7 +38,7 @@ def get_cluster(initialize, leader, members, failover, sync, cluster_config=None history = TimelineHistory(1, '[[1,67197376,"no recovery target specified","' + t + '"]]', [(1, 67197376, 'no recovery target specified', t)]) cluster_config = cluster_config or ClusterConfig(1, {'check_timeline': True}, 1) - return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history) + return Cluster(initialize, cluster_config, leader, 10, members, failover, sync, history, None) def get_cluster_not_initialized_without_leader(cluster_config=None): @@ -66,7 +66,7 @@ def get_cluster_initialized_with_leader(failover=None, sync=None): def get_cluster_initialized_with_only_leader(failover=None, cluster_config=None): leader = get_cluster_initialized_without_leader(leader=True, failover=failover).leader - return get_cluster(True, leader, [leader], failover, None, cluster_config) + return get_cluster(True, leader, [leader.member], failover, None, cluster_config) def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None): @@ -152,13 +152,13 @@ def run_async(self, func, args=()): @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster())) @patch.object(Postgresql, 'is_leader', Mock(return_value=True)) @patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1))) -@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=3)) +@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10)) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) @patch.object(Postgresql, 'controldata', Mock(return_value={ 'Database system identifier': SYSID, 'Database cluster state': 'shut down', 'Latest checkpoint location': '0/12345678'})) -@patch.object(SlotsHandler, 'sync_replication_slots', Mock()) +@patch.object(SlotsHandler, 'load_replication_slots', Mock(side_effect=Exception)) @patch.object(ConfigHandler, 'append_pg_hba', Mock()) @patch.object(ConfigHandler, 'write_pgpass', Mock(return_value={})) @patch.object(ConfigHandler, 'write_recovery_conf', Mock()) @@ -384,13 +384,21 @@ class TestHa(PostgresInit): self.p.is_leader = false self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS') + @patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000)) def test_follow(self): self.ha.cluster.is_unlocked = false self.p.is_leader = false self.assertEqual(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') self.ha.patroni.replicatefrom = "foo" self.p.config.check_recovery_conf = Mock(return_value=(True, False)) + self.ha.cluster.config.data.update({'slots': {'l': {'database': 'a', 'plugin': 'b'}}}) + self.ha.cluster.members[1].data['tags']['replicatefrom'] = 'postgresql0' + self.ha.patroni.nofailover = True self.assertEqual(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') + del self.ha.cluster.config.data['slots'] + self.ha.cluster.config.data.update({'postgresql': {'use_slots': False}}) + self.assertEqual(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') + del self.ha.cluster.config.data['postgresql']['use_slots'] def test_follow_in_pause(self): self.ha.cluster.is_unlocked = false @@ -634,7 +642,7 @@ class TestHa(PostgresInit): # in synchronous_mode consider itself healthy if the former leader is accessible in read-only and ahead of us with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)): self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) - with patch('patroni.postgresql.Postgresql.timeline_wal_position', return_value=(1, 1, 1)): + with patch('patroni.postgresql.Postgresql.last_operation', return_value=1): self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) with patch('patroni.postgresql.Postgresql.replica_cached_timeline', return_value=1): self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) @@ -1110,6 +1118,7 @@ class TestHa(PostgresInit): @patch('psycopg2.connect', psycopg2_connect) def test_permanent_logical_slots_after_promote(self): config = ClusterConfig(1, {'slots': {'l': {'database': 'postgres', 'plugin': 'test_decoding'}}}, 1) + self.p.name = 'other' self.ha.cluster = get_cluster_initialized_without_leader(cluster_config=config) self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader') self.ha.cluster = get_cluster_initialized_without_leader(leader=True, cluster_config=config) @@ -1137,3 +1146,19 @@ class TestHa(PostgresInit): self.ha.has_lock = true self.assertEqual(self.ha.run_cycle(), 'PAUSE: released leader key voluntarily due to the system ID mismatch') + + @patch('psycopg2.connect', psycopg2_connect) + @patch('os.path.exists', Mock(return_value=True)) + @patch('shutil.rmtree', Mock()) + @patch('os.makedirs', Mock()) + @patch('os.open', Mock()) + @patch('os.fsync', Mock()) + @patch('os.close', Mock()) + @patch('os.rename', Mock()) + @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False)) + @patch.object(builtins, 'open', mock_open()) + @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['foo'])) + def test_follow_copy(self): + self.ha.cluster.is_unlocked = false + self.p.is_leader = false + self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots')) diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index aa4493a9..92339b5a 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -16,7 +16,8 @@ def mock_list_namespaced_config_map(*args, **kwargs): metadata = {'resource_version': '1', 'labels': {'f': 'b'}, 'name': 'test-config', 'annotations': {'initialize': '123', 'config': '{}'}} items = [k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))] - metadata.update({'name': 'test-leader', 'annotations': {'optime': '1234', 'leader': 'p-0', 'ttl': '30s'}}) + metadata.update({'name': 'test-leader', + 'annotations': {'optime': '1234x', 'leader': 'p-0', 'ttl': '30s', 'slots': '{'}}) items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) metadata.update({'name': 'test-failover', 'annotations': {'leader': 'p-0'}}) items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) @@ -260,10 +261,6 @@ class TestKubernetesEndpoints(BaseTestKubernetes): self.k._kinds._object_cache['test'].metadata.annotations['leader'] = 'p-1' self.assertFalse(self.k.update_leader('123')) - @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True) - def test_update_leader_with_restricted_access(self): - self.assertIsNotNone(self.k.update_leader('123', True)) - @patch.object(k8s_client.CoreV1Api, 'read_namespaced_endpoints', create=True) @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True) def test__update_leader_with_retry(self, mock_patch, mock_read): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 3a3937d4..479554fe 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,5 +1,4 @@ import datetime -import mock # for the mock.call method, importing it without a namespace breaks python3 import os import psutil import psycopg2 @@ -9,11 +8,10 @@ import time from mock import Mock, MagicMock, PropertyMock, patch, mock_open from patroni.async_executor import CriticalTask -from patroni.dcs import Cluster, ClusterConfig, Member, RemoteMember, SyncState +from patroni.dcs import Cluster, RemoteMember, SyncState from patroni.exceptions import PostgresConnectionException, PatroniException from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE from patroni.postgresql.postmaster import PostmasterProcess -from patroni.postgresql.slots import SlotsHandler from patroni.utils import RetryFailedError from six.moves import builtins from threading import Thread, current_thread @@ -303,29 +301,6 @@ class TestPostgresql(BaseTestPostgresql): m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}}) self.p.follow(m) - @patch.object(Postgresql, 'is_running', Mock(return_value=True)) - def test_sync_replication_slots(self): - self.p.start() - config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'}, - 'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}}, - 'ignore_slots': [{'name': 'blabla'}]}, 1) - cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None) - with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)): - self.p.slots_handler.sync_replication_slots(cluster) - self.p.slots_handler.sync_replication_slots(cluster) - with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')): - self.p.slots_handler.sync_replication_slots(cluster) - with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=True)),\ - patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock: - alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'}) - alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'}) - cluster.members.extend([alias1, alias2]) - self.p.slots_handler.sync_replication_slots(cluster) - self.assertEqual(errorlog_mock.call_count, 5) - ca = errorlog_mock.call_args_list[0][0][1] - self.assertTrue("test-3" in ca, "non matching {0}".format(ca)) - self.assertTrue("test.3" in ca, "non matching {0}".format(ca)) - @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)) def test__query(self): self.assertRaises(PostgresConnectionException, self.p._query, 'blabla') @@ -340,7 +315,7 @@ class TestPostgresql(BaseTestPostgresql): @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT)) def test_is_leader(self): self.assertTrue(self.p.is_leader()) - self.p.reset_cluster_info_state() + self.p.reset_cluster_info_state(None) with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))): self.assertRaises(PostgresConnectionException, self.p.is_leader) @@ -611,7 +586,7 @@ class TestPostgresql(BaseTestPostgresql): def test_pick_sync_standby(self): cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, - SyncState(0, self.me.name, self.leadermem.name), None) + SyncState(0, self.me.name, self.leadermem.name), None, None) mock_cursor = Mock() mock_cursor.fetchone.return_value = ('remote_apply',) @@ -727,10 +702,14 @@ class TestPostgresql(BaseTestPostgresql): @patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))) def test_received_timeline(self): self.p.set_role('standby_leader') - self.p.reset_cluster_info_state() + self.p.reset_cluster_info_state(None) self.assertRaises(PostgresConnectionException, self.p.received_timeline) def test__write_recovery_params(self): self.p.config._write_recovery_params(Mock(), {'pause_at_recovery_target': 'false'}) with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90400)): self.p.config._write_recovery_params(Mock(), {'recovery_target_action': 'PROMOTE'}) + + @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + def test_set_enforce_hot_standby_feedback(self): + self.p.set_enforce_hot_standby_feedback(True) diff --git a/tests/test_raft.py b/tests/test_raft.py index 4b51ebe0..54197df8 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -129,14 +129,19 @@ class TestRaft(unittest.TestCase): 'retry_timeout': 10, 'data_dir': self._TMP}) raft.set_retry_timeout(20) raft.set_ttl(60) + self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}')) self.assertTrue(raft.touch_member('')) self.assertTrue(raft.initialize()) self.assertTrue(raft.cancel_initialization()) self.assertTrue(raft.set_config_value('{}')) self.assertTrue(raft.write_sync_state('foo', 'bar')) - self.assertTrue(raft.update_leader('1')) self.assertTrue(raft.manual_failover('foo', 'bar')) raft.get_cluster() + self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}')) + raft.get_cluster() + self.assertTrue(raft.update_leader('1')) + self.assertTrue(raft._sync_obj.set(raft.status_path, '{')) + raft.get_cluster() self.assertTrue(raft.delete_sync_state()) self.assertTrue(raft.delete_leader()) self.assertTrue(raft.set_history_value('')) diff --git a/tests/test_rewind.py b/tests/test_rewind.py index cb14ce9d..c2607e3f 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -93,7 +93,7 @@ class TestRewind(BaseTestPostgresql): self.r.rewind_or_reinitialize_needed_and_possible(self.leader) with patch.object(Postgresql, 'is_running', Mock(return_value=True)): - with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1,), Exception])): + with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None), Exception])): self.r.rewind_or_reinitialize_needed_and_possible(self.leader) @patch.object(CancellableSubprocess, 'call', mock_cancellable_call) diff --git a/tests/test_slots.py b/tests/test_slots.py new file mode 100644 index 00000000..e17c4779 --- /dev/null +++ b/tests/test_slots.py @@ -0,0 +1,124 @@ +import mock +import os +import psycopg2 +import unittest + + +from mock import Mock, PropertyMock, patch + +from patroni.dcs import Cluster, ClusterConfig, Member +from patroni.postgresql import Postgresql +from patroni.postgresql.slots import SlotsHandler, fsync_dir + +from . import BaseTestPostgresql, psycopg2_connect, MockCursor + + +@patch('subprocess.call', Mock(return_value=0)) +@patch('psycopg2.connect', psycopg2_connect) +@patch.object(Postgresql, 'is_running', Mock(return_value=True)) +class TestSlotsHandler(BaseTestPostgresql): + + @patch('subprocess.call', Mock(return_value=0)) + @patch('os.rename', Mock()) + @patch('patroni.postgresql.CallbackExecutor', Mock()) + @patch.object(Postgresql, 'get_major_version', Mock(return_value=130000)) + @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + def setUp(self): + super(TestSlotsHandler, self).setUp() + self.s = self.p.slots_handler + self.p.start() + + def test_sync_replication_slots(self): + config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'}, + 'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}}, + 'ignore_slots': [{'name': 'blabla'}]}, 1) + cluster = Cluster(True, config, self.leader, 0, + [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10}) + with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)): + self.s.sync_replication_slots(cluster, False) + self.p.set_role('standby_leader') + self.s.sync_replication_slots(cluster, False) + self.p.set_role('replica') + with patch.object(Postgresql, 'is_leader', Mock(return_value=False)): + self.s.sync_replication_slots(cluster, False) + self.p.set_role('master') + with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')): + self.s.sync_replication_slots(cluster, False) + with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=True)),\ + patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock: + alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'}) + alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'}) + cluster.members.extend([alias1, alias2]) + self.s.sync_replication_slots(cluster, False) + self.assertEqual(errorlog_mock.call_count, 5) + ca = errorlog_mock.call_args_list[0][0][1] + self.assertTrue("test-3" in ca, "non matching {0}".format(ca)) + self.assertTrue("test.3" in ca, "non matching {0}".format(ca)) + with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)): + self.s.sync_replication_slots(cluster, False) + + def test_process_permanent_slots(self): + config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}, + 'ignore_slots': [{'name': 'blabla'}]}, 1) + cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], None, None, None, None) + + self.s.sync_replication_slots(cluster, False) + with patch.object(Postgresql, '_query') as mock_query: + self.p.reset_cluster_info_state(None) + mock_query.return_value.fetchone.return_value = ( + 1, 0, 0, 0, 0, 0, 0, 0, 0, + [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b", + "confirmed_flush_lsn": 12345, "catalog_xmin": 105}]) + self.assertEqual(self.p.slots(), {'ls': 12345}) + + self.p.reset_cluster_info_state(None) + mock_query.return_value.fetchone.return_value = ( + 1, 0, 0, 0, 0, 0, 0, 0, 0, + [{"slot_name": "ls", "type": "logical", "datoid": 6, "plugin": "b", + "confirmed_flush_lsn": 12345, "catalog_xmin": 105}]) + self.assertEqual(self.p.slots(), {}) + + @patch.object(Postgresql, 'is_leader', Mock(return_value=False)) + def test__ensure_logical_slots_replica(self): + self.p.set_role('replica') + config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1) + cluster = Cluster(True, config, self.leader, 0, + [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346}) + self.assertEqual(self.s.sync_replication_slots(cluster, False), []) + self.s._schedule_load_slots = False + with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)): + self.assertEqual(self.s.sync_replication_slots(cluster, False), []) + cluster.slots['ls'] = 'a' + self.assertEqual(self.s.sync_replication_slots(cluster, False), []) + with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True): + self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls']) + + @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)) + def test_copy_logical_slots(self): + self.s.copy_logical_slots(self.leader, ['foo']) + + @patch.object(Postgresql, 'stop', Mock(return_value=True)) + @patch.object(Postgresql, 'start', Mock(return_value=True)) + @patch.object(Postgresql, 'is_leader', Mock(return_value=False)) + def test_check_logical_slots_readiness(self): + self.s.copy_logical_slots(self.leader, ['ls']) + config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1) + cluster = Cluster(True, config, self.leader, 0, + [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345}) + self.assertEqual(self.s.sync_replication_slots(cluster, False), []) + with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True): + self.s.check_logical_slots_readiness(cluster, False, None) + + @patch.object(Postgresql, 'stop', Mock(return_value=True)) + @patch.object(Postgresql, 'start', Mock(return_value=True)) + @patch.object(Postgresql, 'is_leader', Mock(return_value=False)) + def test_on_promote(self): + self.s.copy_logical_slots(self.leader, ['ls']) + self.s.on_promote() + + @unittest.skipIf(os.name == 'nt', "Windows not supported") + @patch('os.open', Mock()) + @patch('os.close', Mock()) + @patch('os.fsync', Mock(side_effect=OSError)) + def test_fsync_dir(self): + self.assertRaises(OSError, fsync_dir, 'foo') diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 8d00894a..ef368826 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -30,7 +30,9 @@ class MockKazooClient(Mock): def get(self, path, watch=None): if not isinstance(path, six.string_types): raise TypeError("Invalid type for 'path' (string expected)") - if path == '/no_node': + if path == '/broken/status': + return (b'{', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0)) + elif path in ('/no_node', '/legacy/status'): raise NoNodeError elif '/members/' in path: return ( @@ -45,6 +47,8 @@ class MockKazooClient(Mock): return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) elif path.endswith('/initialize'): return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) + elif path.endswith('/status'): + return (b'{"optime":500,"slots":{"ls":1234567}}', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0)) return (b'', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) @staticmethod @@ -147,6 +151,10 @@ class TestZooKeeper(unittest.TestCase): def test__inner_load_cluster(self): self.zk._base_path = self.zk._base_path.replace('test', 'bla') self.zk._inner_load_cluster() + self.zk._base_path = self.zk._base_path = '/broken' + self.zk._inner_load_cluster() + self.zk._base_path = self.zk._base_path = '/legacy' + self.zk._inner_load_cluster() self.zk._base_path = self.zk._base_path = '/no_node' self.zk._inner_load_cluster() @@ -156,11 +164,11 @@ class TestZooKeeper(unittest.TestCase): self.assertIsInstance(cluster.leader, Leader) self.zk.touch_member({'foo': 'foo'}) self.zk._name = 'bar' - self.zk.optime_watcher(None) + self.zk.status_watcher(None) with patch.object(ZooKeeper, 'get_node', Mock(side_effect=Exception)): self.zk.get_cluster() cluster = self.zk.get_cluster() - self.assertEqual(cluster.last_leader_operation, 500) + self.assertEqual(cluster.last_lsn, 500) def test_delete_leader(self): self.assertTrue(self.zk.delete_leader()) @@ -203,10 +211,10 @@ class TestZooKeeper(unittest.TestCase): self.zk.take_leader() def test_update_leader(self): - self.assertTrue(self.zk.update_leader(None)) + self.assertTrue(self.zk.update_leader(12345)) def test_write_leader_optime(self): - self.zk.last_leader_operation = '0' + self.zk.last_lsn = '0' self.zk.write_leader_optime('1') with patch.object(MockKazooClient, 'create_async', Mock()): self.zk.write_leader_optime('1') @@ -221,7 +229,7 @@ class TestZooKeeper(unittest.TestCase): def test_watch(self): self.zk.watch(None, 0) self.zk.event.isSet = Mock(return_value=True) - self.zk._fetch_optime = False + self.zk._fetch_status = False self.zk.watch(None, 0) def test__kazoo_connect(self):