From 71863cedcb4e51cf0e3077e8f56cd388055f640c Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Thu, 14 Sep 2023 18:34:45 +0200 Subject: [PATCH 1/8] Always store CMDLINE_OPTIONS config values as int (#2861) --- patroni/config.py | 6 ++++-- patroni/postgresql/citus.py | 2 +- tests/test_bootstrap.py | 3 ++- tests/test_config.py | 25 ++++++++++++++++++++++++- tests/test_postgresql.py | 2 +- 5 files changed, 32 insertions(+), 6 deletions(-) diff --git a/patroni/config.py b/patroni/config.py index 65991b0e..1650934d 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -16,6 +16,7 @@ from .dcs import ClusterConfig, Cluster from .exceptions import ConfigParseError from .file_perm import pg_perm from .postgresql.config import ConfigHandler +from .validator import IntValidator from .utils import deep_compare, parse_bool, parse_int, patch_config logger = logging.getLogger(__name__) @@ -487,8 +488,9 @@ class Config(object): if name not in ConfigHandler.CMDLINE_OPTIONS: pg_params[name] = value elif not is_local: - if ConfigHandler.CMDLINE_OPTIONS[name][1](value): - pg_params[name] = value + validator = ConfigHandler.CMDLINE_OPTIONS[name][1] + if validator(value): + pg_params[name] = int(value) if isinstance(validator, IntValidator) else value else: logger.warning("postgresql parameter %s=%s failed validation, defaulting to %s", name, value, ConfigHandler.CMDLINE_OPTIONS[name][0]) diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index 09f77f2b..8ca2790e 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -397,7 +397,7 @@ class CitusHandler(Thread): parameters['shared_preload_libraries'] = ','.join(['citus'] + shared_preload_libraries) # if not explicitly set Citus overrides max_prepared_transactions to max_connections*2 - if parameters.get('max_prepared_transactions') == 0: + if parameters['max_prepared_transactions'] == 0: parameters['max_prepared_transactions'] = parameters['max_connections'] * 2 # Resharding in Citus implemented using logical replication diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index c922fcae..9ac6aa68 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -238,7 +238,8 @@ class TestBootstrap(BaseTestPostgresql): self.p.reload_config({'authentication': {'superuser': {'username': 'p', 'password': 'p'}, 'replication': {'username': 'r', 'password': 'r'}, 'rewind': {'username': 'rw', 'password': 'rw'}}, - 'listen': '*', 'retry_timeout': 10, 'parameters': {'wal_level': '', 'hba_file': 'foo'}}) + 'listen': '*', 'retry_timeout': 10, + 'parameters': {'wal_level': '', 'hba_file': 'foo', 'max_prepared_transactions': 10}}) with patch.object(Postgresql, 'major_version', PropertyMock(return_value=110000)), \ patch.object(Postgresql, 'restart', Mock()) as mock_restart: self.b.post_bootstrap({}, task) diff --git a/tests/test_config.py b/tests/test_config.py index cf798d00..bd8d0a90 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -3,6 +3,7 @@ import sys import unittest import io +from copy import deepcopy from mock import MagicMock, Mock, patch from patroni.config import Config, ConfigParseError @@ -22,7 +23,7 @@ class TestConfig(unittest.TestCase): self.assertFalse(self.config.set_dynamic_configuration({'foo': 'bar'})) self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}, 'postgresql': { 'parameters': {'cluster_name': 1, 'hot_standby': 1, 'wal_keep_size': 1, - 'track_commit_timestamp': 1, 'wal_level': 1}}})) + 'track_commit_timestamp': 1, 'wal_level': 1, 'max_connections': '100'}}})) def test_reload_local_configuration(self): os.environ.update({ @@ -149,3 +150,25 @@ class TestConfig(unittest.TestCase): @patch('os.path.isdir', Mock(return_value=False)) def test_invalid_path(self): self.assertRaises(ConfigParseError, Config, 'postgres0') + + def test__process_postgresql_parameters(self): + expected_params = { + 'f.oo': 'bar', # not in ConfigHandler.CMDLINE_OPTIONS + 'max_connections': 100, # IntValidator + 'wal_level': 'hot_standby', # EnumValidator + } + input_params = deepcopy(expected_params) + + input_params['max_connections'] = '100' + self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params) + + expected_params['f.oo'] = input_params['f.oo'] = '100' + self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params) + + input_params['wal_level'] = 'cold_standby' + expected_params.pop('wal_level') + self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params) + + input_params['max_connections'] = 10 + expected_params.pop('max_connections') + self.assertEqual(self.config._process_postgresql_parameters(input_params), expected_params) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 9ab3f52d..e2fad6ba 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -688,7 +688,7 @@ class TestPostgresql(BaseTestPostgresql): self.assertIsNone(self.p.wait_for_startup()) def test_get_server_parameters(self): - config = {'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'} + config = {'parameters': {'wal_level': 'hot_standby', 'max_prepared_transactions': 100}, 'listen': '0'} self.p._global_config = GlobalConfig({'synchronous_mode': True}) self.p.config.get_server_parameters(config) self.p._global_config = GlobalConfig({'synchronous_mode': True, 'synchronous_mode_strict': True}) From 75dbe4ff96a7443fde4fe3d2ddc7492ec2710530 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 14 Sep 2023 19:36:26 +0200 Subject: [PATCH 2/8] Update supported Postgres versions (#2857) --- .github/workflows/install_deps.py | 4 ++-- .github/workflows/mapping.py | 2 +- README.rst | 2 +- docs/index.rst | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py index 29acb701..ea0c9d1b 100644 --- a/.github/workflows/install_deps.py +++ b/.github/workflows/install_deps.py @@ -45,8 +45,8 @@ def install_packages(what): packages['exhibitor'] = packages['zookeeper'] packages = packages.get(what, []) ver = versions.get(what) - if float(ver) >= 15: - packages += ['postgresql-{0}-citus-11.2'.format(ver)] + if float(ver) == 15: + packages += ['postgresql-{0}-citus-12.0'.format(ver)] subprocess.call(['sudo', 'apt-get', 'update', '-y']) return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev'] + packages) diff --git a/.github/workflows/mapping.py b/.github/workflows/mapping.py index 397a4cc7..f75efec4 100644 --- a/.github/workflows/mapping.py +++ b/.github/workflows/mapping.py @@ -1 +1 @@ -versions = {'etcd': '9.6', 'etcd3': '14', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '15'} +versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '15'} diff --git a/README.rst b/README.rst index f8f187e4..decc868f 100644 --- a/README.rst +++ b/README.rst @@ -12,7 +12,7 @@ Patroni is a template for high availability (HA) PostgreSQL solutions using Pyth We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely. -Currently supported PostgreSQL versions: 9.3 to 15. +Currently supported PostgreSQL versions: 9.3 to 16. **Note to Citus users**: Starting from 3.0 Patroni nicely integrates with the `Citus `__ database extension to Postgres. Please check the `Citus support page `__ in the Patroni documentation for more info about how to use Patroni high availability together with a Citus distributed cluster. diff --git a/docs/index.rst b/docs/index.rst index 440f2278..c8f94aaf 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -10,7 +10,7 @@ Patroni is a template for high availability (HA) PostgreSQL solutions using Pyth We call Patroni a "template" because it is far from being a one-size-fits-all or plug-and-play replication system. It will have its own caveats. Use wisely. There are many ways to run high availability with PostgreSQL; for a list, see the `PostgreSQL Documentation `__. -Currently supported PostgreSQL versions: 9.3 to 15. +Currently supported PostgreSQL versions: 9.3 to 16. **Note to Citus users**: Starting from 3.0 Patroni nicely integrates with the `Citus `__ database extension to Postgres. Please check the :ref:`Citus support page ` in the Patroni documentation for more info about how to use Patroni high availability together with a Citus distributed cluster. From 5a504e67c132b979a4c687fd87f0f589884167f5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 15 Sep 2023 11:32:49 +0200 Subject: [PATCH 3/8] Don't rely on pg_stat_wal_receiver when deciding on pg_rewind (#2863) As was reported by @ants on Slack it could happen that `received_tli` is ahead of replayed timeline, therefore we should stop using it when deciding on pg_rewind if postgres is running and use only `IDENTIFY_SYSTEM` via replication connection. --- patroni/postgresql/rewind.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 73bccb44..6e1aab88 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -158,7 +158,7 @@ class Rewind(object): def _get_local_timeline_lsn(self) -> Tuple[Optional[bool], Optional[int], Optional[int]]: if self._postgresql.is_running(): # if postgres is running - get timeline from replication connection in_recovery = True - timeline = self._postgresql.received_timeline() or self._postgresql.get_replica_timeline() + timeline = self._postgresql.get_replica_timeline() lsn = self._postgresql.replayed_location() else: # otherwise analyze pg_controldata output in_recovery, timeline, lsn = self._get_local_timeline_lsn_from_controldata() From 25ceb6825741fb3ac6f9e535af4f3336301edbb0 Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Mon, 18 Sep 2023 15:30:35 +0200 Subject: [PATCH 4/8] Fix k8s dockerfiles (#2870) - Allow pip to modify an EXTERNALLY-MANAGED Python installation by passing --break-system-packages - Build Citus for arm64 - Don't use PG_MAJOR argument --- kubernetes/Dockerfile | 4 ++-- kubernetes/Dockerfile.citus | 27 +++++++++++++++++++++------ 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/kubernetes/Dockerfile b/kubernetes/Dockerfile index 2fa25f73..29a683bd 100644 --- a/kubernetes/Dockerfile +++ b/kubernetes/Dockerfile @@ -9,8 +9,8 @@ RUN export DEBIAN_FRONTEND=noninteractive \ | xargs apt-get install -y vim-tiny curl jq locales git python3-pip python3-wheel \ ## Make sure we have a en_US.UTF-8 locale available && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \ - && pip3 install setuptools \ - && pip3 install 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \ + && pip3 install --break-system-packages setuptools \ + && pip3 install --break-system-packages 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \ && PGHOME=/home/postgres \ && mkdir -p $PGHOME \ && chown postgres $PGHOME \ diff --git a/kubernetes/Dockerfile.citus b/kubernetes/Dockerfile.citus index 1ae242bf..f9564521 100644 --- a/kubernetes/Dockerfile.citus +++ b/kubernetes/Dockerfile.citus @@ -10,12 +10,24 @@ RUN export DEBIAN_FRONTEND=noninteractive \ | xargs apt-get install -y busybox vim-tiny curl jq less locales git python3-pip python3-wheel lsb-release \ ## Make sure we have a en_US.UTF-8 locale available && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \ - && echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \ - && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \ - && apt-get update -y \ - && apt-get -y install postgresql-$PG_MAJOR-citus-11.3 \ - && pip3 install setuptools \ - && pip3 install 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \ + && if [ $(dpkg --print-architecture) = 'arm64' ]; then \ + apt-get install -y postgresql-server-dev-15 \ + gcc make autoconf \ + libc6-dev flex libcurl4-gnutls-dev \ + libicu-dev libkrb5-dev liblz4-dev \ + libpam0g-dev libreadline-dev libselinux1-dev\ + libssl-dev libxslt1-dev libzstd-dev uuid-dev \ + && git clone -b "main" https://github.com/citusdata/citus.git \ + && MAKEFLAGS="-j $(grep -c ^processor /proc/cpuinfo)" \ + && cd citus && ./configure && make install && cd ../ && rm -rf /citus; \ + else \ + echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \ + && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \ + && apt-get update -y \ + && apt-get -y install postgresql-15-citus-12.0; \ + fi \ + && pip3 install --break-system-packages setuptools \ + && pip3 install --break-system-packages 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \ && PGHOME=/home/postgres \ && mkdir -p $PGHOME \ && chown postgres $PGHOME \ @@ -26,6 +38,9 @@ RUN export DEBIAN_FRONTEND=noninteractive \ && chmod 664 /etc/passwd \ # Clean up && apt-get remove -y git python3-pip python3-wheel \ + postgresql-server-dev-15 gcc make autoconf \ + libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \ + libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \ && apt-get autoremove -y \ && apt-get clean -y \ && rm -rf /var/lib/apt/lists/* /root/.cache From 28b9d3d2d9564bcfa52c4ecf6720d1005549c832 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 19 Sep 2023 10:32:35 +0200 Subject: [PATCH 5/8] Bump pyright version (#2871) and fix all reported issues. We aren't sticking to the latest version this time because it has [a bug](https://github.com/microsoft/pyright/issues/5968). --- .github/workflows/tests.yaml | 2 +- patroni/dcs/etcd3.py | 2 +- patroni/dcs/kubernetes.py | 2 +- patroni/dcs/zookeeper.py | 2 +- patroni/ha.py | 4 ++-- patroni/postgresql/__init__.py | 4 ++-- patroni/postgresql/config.py | 7 ++----- patroni/postgresql/slots.py | 2 +- 8 files changed, 11 insertions(+), 14 deletions(-) diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index bc6a1be3..8b73c6e7 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -173,7 +173,7 @@ jobs: - uses: jakebailey/pyright-action@v1 with: - version: 1.1.320 + version: 1.1.326 docs: runs-on: ubuntu-latest diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 0a71caa0..5cbd813f 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -205,7 +205,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None: self._token = None - self._cluster_version: Tuple[int] = tuple() + self._cluster_version: Tuple[int, ...] = tuple() super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl) try: diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 3be7cee3..d4158368 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -836,7 +836,7 @@ class Kubernetes(AbstractDCS): self._api.configure_timeouts(self.loop_wait, self._retry.deadline, self.ttl) # retriable_http_codes supposed to be either int, list of integers or comma-separated string with integers. - retriable_http_codes = config.get('retriable_http_codes', []) + retriable_http_codes: Union[str, List[Union[str, int]]] = config.get('retriable_http_codes', []) if not isinstance(retriable_http_codes, list): retriable_http_codes = [c.strip() for c in str(retriable_http_codes).split(',')] diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index e4af1b1f..e3cc1d69 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -90,7 +90,7 @@ class ZooKeeper(AbstractDCS): def __init__(self, config: Dict[str, Any]) -> None: super(ZooKeeper, self).__init__(config) - hosts = config.get('hosts', []) + hosts: Union[str, List[str]] = config.get('hosts', []) if isinstance(hosts, list): hosts = ','.join(hosts) diff --git a/patroni/ha.py b/patroni/ha.py index a8274ee3..d5d6e16b 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -772,13 +772,13 @@ class Ha(object): if cluster_history: self.dcs.set_history_value('[]') elif not cluster_history or cluster_history[-1][0] != primary_timeline - 1 or len(cluster_history[-1]) != 5: - cluster_history = {line[0]: line for line in cluster_history} + cluster_history_dict: Dict[int, List[Any]] = {line[0]: list(line) for line in cluster_history} history: List[List[Any]] = list(map(list, self.state_handler.get_history(primary_timeline))) if self.cluster.config: history = history[-self.cluster.config.max_timelines_history:] for line in history: # enrich current history with promotion timestamps stored in DCS - cluster_history_line = list(cluster_history.get(line[0], [])) + cluster_history_line = cluster_history_dict.get(line[0], []) if len(line) == 3 and len(cluster_history_line) >= 4 and cluster_history_line[1] == line[1]: line.append(cluster_history_line[3]) if len(cluster_history_line) == 5: diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index a37e15e1..9bd53dac 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -568,7 +568,7 @@ class Postgresql(object): r'lsn: ([0-9A-Fa-f]+/[0-9A-Fa-f]+), prev ([0-9A-Fa-f]+/[0-9A-Fa-f]+), ' r'.*?desc: (.+)', out.decode('utf-8')) if match: - return match.groups() + return match.group(1), match.group(2), match.group(3), match.group(4) return None, None, None, None def latest_checkpoint_location(self) -> Optional[int]: @@ -1023,7 +1023,7 @@ class Postgresql(object): return None, None @contextmanager - def get_replication_connection_cursor(self, host: Optional[str] = None, port: int = 5432, + def get_replication_connection_cursor(self, host: Optional[str] = None, port: Union[int, str] = 5432, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]: conn_kwargs = self.config.replication.copy() conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'), diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 315bf8c7..a918a2cf 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -1026,17 +1026,14 @@ class ConfigHandler(object): # "notify" connection_pool about the "new" local connection address self._postgresql.connection_pool.conn_kwargs = local_conn_kwargs - def _get_pg_settings( - self, names: Collection[str] - ) -> Dict[str, Tuple[str, str, Optional[str], str, str, Optional[str]]]: + def _get_pg_settings(self, names: Collection[str]) -> Dict[Any, Tuple[Any, ...]]: return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context, sourcefile' + ' FROM pg_catalog.pg_settings ' + ' WHERE pg_catalog.lower(name) = ANY(%s)'), [n.lower() for n in names])} @staticmethod - def _handle_wal_buffers(old_values: Dict[str, Tuple[str, str, Optional[str], str, str, Optional[str]]], - changes: CaseInsensitiveDict) -> None: + def _handle_wal_buffers(old_values: Dict[Any, Tuple[Any, ...]], changes: CaseInsensitiveDict) -> None: wal_block_size = parse_int(old_values['wal_block_size'][1]) or 8192 wal_segment_size = old_values['wal_segment_size'] wal_segment_unit = parse_int(wal_segment_size[2], 'B') or 8192 \ diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 7391f543..f6090a3b 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -313,7 +313,7 @@ class SlotsHandler: ' true AS dropped FROM slots WHERE not active) ' 'SELECT active, COALESCE(dropped, false) FROM slots' ' FULL OUTER JOIN dropped ON true'), name) - return rows[0] if rows else (False, False) + return (rows[0][0], rows[0][1]) if rows else (False, False) def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any], paused: bool) -> None: """Compare required slots and configured as permanent slots with those found, dropping extraneous ones. From 66bdb1ae12f5eaa6fc20386fa50e046acde86a98 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 20 Sep 2023 12:00:18 +0200 Subject: [PATCH 6/8] Release v3.1.1 (#2872) * Bump version * Update release notes * Update contributing guidelines and tox.ini (include v16) * Enable tests for `REL*` branches --- .github/workflows/tests.yaml | 1 + docs/contributing_guidelines.rst | 4 +-- docs/releases.rst | 58 ++++++++++++++++++++++++++++++++ patroni/version.py | 2 +- tox.ini | 7 ++-- 5 files changed, 66 insertions(+), 6 deletions(-) diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 8b73c6e7..aba3e751 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -5,6 +5,7 @@ on: push: branches: - master + - 'REL_[0-9]+_[0-9]+' env: CODACY_PROJECT_TOKEN: ${{ secrets.CODACY_PROJECT_TOKEN }} diff --git a/docs/contributing_guidelines.rst b/docs/contributing_guidelines.rst index 48710868..05d98987 100644 --- a/docs/contributing_guidelines.rst +++ b/docs/contributing_guidelines.rst @@ -143,12 +143,12 @@ If you want to disable this facility set the env var `OPEN_CMD` to the `:` no-op Behave tests ^^^^^^^^^^^^ -Behave tests with `-m behave` will build docker images based on PG_MAJOR version 11 through 15 and then run all +Behave tests with `-m behave` will build docker images based on PG_MAJOR version 11 through 16 and then run all behave tests. This can take quite a long time to run so you might want to limit the scope to a select version of Postgres or to a specific feature set or steps. To specify the version of postgres include the full name of the dependent image build env that you want and then the -behave env name. For instance if you want Postgres 15 use: +behave env name. For instance if you want Postgres 14 use: .. code-block:: bash diff --git a/docs/releases.rst b/docs/releases.rst index 7a05bf7e..5d4af2cc 100644 --- a/docs/releases.rst +++ b/docs/releases.rst @@ -3,6 +3,64 @@ Release notes ============= +Version 3.1.1 +------------- + +**Bugfixes** + +- Reset failsafe state on promote (ChenChangAo) + + If switchover/failover happened shortly after failsafe mode had been activated, the newly promoted primary was demoting itself after failsafe becomes inactive. + +- Silence useless warnings in ``patronictl`` (Alexander Kukushkin) + + If ``patronictl`` uses the same patroni.yaml file as Patroni and can access ``PGDATA`` directory it might have been showing annoying warnings about incorrect values in the global configuration. + +- Explicitly enable synchronous mode for a corner case (Alexander Kukushkin) + + Synchronous mode effectively was never activated if there are no replicas streaming from the primary. + +- Fixed bug with ``0`` integer values validation (Israel Barth Rubio) + + In most cases, it didn't cause any issues, just warnings. + +- Don't return logical slots for standby cluster (Alexander Kukushkin) + + Patroni can't create logical replication slots in the standby cluster, thus they should be ignored if they are defined in the global configuration. + +- Avoid showing docstring in ``patronictl --help`` output (Israel Barth Rubio) + + The ``click`` module needs to get a special hint for that. + +- Fixed bug with ``kubernetes.standby_leader_label_value`` (Alexander Kukushkin) + + This feature effectively never worked. + +- Returned cluster system identifier to the ``patronictl list`` output (Polina Bungina) + + The problem was introduced while implementing the support for Citus, where we need to hide the identifier because it is different for coordinator and all workers. + +- Override ``write_leader_optime`` method in Kubernetes implementation (Alexander Kukushkin) + + The method is supposed to write shutdown LSN to the leader Endpoint/ConfigMap when there are no healthy replicas available to become the new primary. + +- Don't start stopped postgres in pause (Alexander Kukushkin) + + Due to a race condition, Patroni was falsely assuming that the standby should be restarted because some recovery parameters (``primary_conninfo`` or similar) were changed. + +- Fixed bug in ``patronictl query`` command (Israel Barth Rubio) + + It didn't work when only ``-m`` argument was provided or when none of ``-r`` or ``-m`` were provided. + +- Properly treat integer parameters that are used in the command line to start postgres (Polina Bungina) + + If values are supplied as strings and not casted to integer it was resulting in an incorrect calculation of ``max_prepared_transactions`` based on ``max_connections`` for Citus clusters. + +- Don't rely on ``pg_stat_wal_receiver`` when deciding on ``pg_rewind`` (Alexander Kukushkin) + + It could happen that ``received_tli`` reported by ``pg_stat_wal_recevier`` is ahead of the actual replayed timeline, while the timeline reported by ``DENTIFY_SYSTEM`` via replication connection is always correct. + + Version 3.1.0 ------------- diff --git a/patroni/version.py b/patroni/version.py index 87eff52e..ff98d3c0 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -2,4 +2,4 @@ :var __version__: the current Patroni version. """ -__version__ = '3.1.0' +__version__ = '3.1.1' diff --git a/tox.ini b/tox.ini index b145f483..bf66caef 100644 --- a/tox.ini +++ b/tox.ini @@ -6,6 +6,7 @@ postgres_matrix = pg13: PG_MAJOR = 13 pg14: PG_MAJOR = 14 pg15: PG_MAJOR = 15 + pg16: PG_MAJOR = 16 psycopg_deps = py{37,38,39,310,311}-{lin,win}: psycopg[binary] mac: psycopg2-binary @@ -106,7 +107,7 @@ description = Reformat code with black deps = black commands = black {posargs:patroni tests} -[testenv:pg{12,13,14,15}-docker-build] +[testenv:pg{12,13,14,15,16}-docker-build] description = Build docker containers needed for testing labels = behave @@ -124,7 +125,7 @@ commands = --file features/Dockerfile allowlist_externals = docker -[testenv:pg{12,13,14,15}-docker-behave-{etcd}-{lin,mac}] +[testenv:pg{12,13,14,15,16}-docker-behave-{etcd}-{lin,mac}] description = Run behaviour tests in patroni-dev docker container setenv = etcd: DCS=etcd @@ -133,7 +134,7 @@ setenv = labels = behave depends = - pg{11,12,13,14,15}-docker-build + pg{11,12,13,14,15,16}-docker-build # There's a bug which affects calling multiple envs on the command line # This should be a valid command: tox -e 'py{36,37,38,39,310,311}-behave-{env:DCS}-lin' From 18d9cb1124a8e7b4002b01d64a46075a015e232a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 20 Sep 2023 14:59:15 +0200 Subject: [PATCH 7/8] Stick with sphinx_rtd_theme (#2873) by default they are using something else --- docs/conf.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/conf.py b/docs/conf.py index f94b9660..950fbc91 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -112,10 +112,10 @@ todo_include_todos = True # a list of builtin themes. # +html_theme = 'sphinx_rtd_theme' on_rtd = os.environ.get('READTHEDOCS', None) == 'True' if not on_rtd: # only import and set the theme if we're building docs locally import sphinx_rtd_theme - html_theme = 'sphinx_rtd_theme' html_theme_path = [sphinx_rtd_theme.get_html_theme_path()] # Theme options are theme-specific and customize the look and feel of a theme From bc15813de00c1932dd510ea8c1af257b45732c77 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 20 Sep 2023 16:50:37 +0200 Subject: [PATCH 8/8] Permanent physical slots on standby nodes (#2852) Create permanent physical replication slots on standby nodes and use `pg_replication_slot_advance()` function to move them forward. The `restart_lsn` is advanced based on values stored in the `/status` key by the primary node. When slot is created on a replica it could be ahead the same slot on the primary and therefore there is some period of time when it doesn't protect WAL files from being recycled. --- docs/dynamic_configuration.rst | 17 +++++- features/ignored_slots.feature | 24 ++++---- features/permanent_slots.feature | 34 +++++++++++ features/standby_cluster.feature | 3 - features/steps/slots.py | 61 +++++++++++++++----- patroni/dcs/__init__.py | 96 ++++++++++++++++++++++---------- patroni/dcs/zookeeper.py | 2 +- patroni/postgresql/__init__.py | 25 +++++---- patroni/postgresql/slots.py | 54 ++++++++++++------ tests/__init__.py | 10 ++-- tests/test_slots.py | 32 +++++++++-- 11 files changed, 256 insertions(+), 102 deletions(-) create mode 100644 features/permanent_slots.feature diff --git a/docs/dynamic_configuration.rst b/docs/dynamic_configuration.rst index f5bb4ee5..7f04ce33 100644 --- a/docs/dynamic_configuration.rst +++ b/docs/dynamic_configuration.rst @@ -46,7 +46,7 @@ In order to change the dynamic configuration you can use either ``patronictl edi - **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. Permanent slots that don't exist will be created by Patroni. The physical slots are maintained only in the current primary. 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+. +- **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. With PostgreSQL 11 onwards permanent physical slots are created on all nodes and their position is advanced every **loop_wait** seconds. For PostgreSQL versions older than 11 permanent physical replication slots are maintained only on the current primary. 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 replication slots requires **postgresql.use_slots** to be set to ``true``. If there are permanent logical replication slots defined Patroni will 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 the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster ` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni. @@ -80,4 +80,17 @@ Note: **slots** is a hashmap while **ignore_slots** is an array. For example: plugin: test_decoding - name: ignored_physical_slot_name type: physical - ... \ No newline at end of file + ... + +Note: if cluster topology is static (fixed number of nodes that never change their names) you can configure permanent physical replication slots with names corresponding to names of nodes to avoid recycling of WAL files while replica is temporary down: + +.. code:: YAML + + slots: + node_name1: + type: physical + node_name2: + type: physical + node_name3: + type: physical + ... diff --git a/features/ignored_slots.feature b/features/ignored_slots.feature index 08a6dda5..e0c53ea3 100644 --- a/features/ignored_slots.feature +++ b/features/ignored_slots.feature @@ -25,10 +25,10 @@ Feature: ignored slots # but Patroni can actually end up dropping them almost immediately, so it's helpful # to verify they exist before we begin testing whether they persist through failover # cycles. - Then postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin + Then postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds When I start postgres0 Then "members/postgres0" key in DCS has role=replica after 10 seconds @@ -46,16 +46,16 @@ Feature: ignored slots And "members/postgres1" key in DCS has role=replica after 10 seconds # give Patroni time to sync replication slots And I sleep for 2 seconds - And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds And postgres1 does not have a logical replication slot named dummy_slot # 3. After a failover the server (now a primary) still has the slot. When I shut down postgres0 Then "members/postgres1" key in DCS has role=master after 10 seconds - And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin - And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds diff --git a/features/permanent_slots.feature b/features/permanent_slots.feature new file mode 100644 index 00000000..656e6ade --- /dev/null +++ b/features/permanent_slots.feature @@ -0,0 +1,34 @@ +Feature: permanent slots + Scenario: check that physical permanent slots are created + Given I start postgres0 + Then postgres0 is a leader after 10 seconds + And there is a non empty initialize key in DCS after 15 seconds + When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_physical": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}} + Then I receive a response code 200 + And Response on GET http://127.0.0.1:8008/config contains slots after 10 seconds + Then postgres0 has a physical replication slot named test_physical after 10 seconds + And I start postgres1 + + @slot-advance + Scenario: check that logical permanent slots are created + Given I run patronictl.py restart batman postgres0 --force + And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}} + Then postgres0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds + + @slot-advance + Scenario: check that permanent slots are created on the replica + Given postgres1 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds + Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds + And postgres1 has a physical replication slot named test_physical after 2 seconds + + @slot-advance + Scenario: check that permanent slots are advanced on the replica + Given I add the table replicate_me to postgres0 + And I get all changes from physical slot test_physical on postgres0 + When I get all changes from logical slot test_logical on postgres0 + Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds + And Physical slot test_physical is in sync between postgres0 and postgres1 after 10 seconds + + Scenario: check permanent physical replication slot after failover + Given I shut down postgres0 + Then postgres1 has a physical replication slot named test_physical after 10 seconds diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature index 850c7970..7ff07679 100644 --- a/features/standby_cluster.feature +++ b/features/standby_cluster.feature @@ -22,9 +22,6 @@ Feature: standby cluster 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 diff --git a/features/steps/slots.py b/features/steps/slots.py index 8a742b1c..f4a3cfa5 100644 --- a/features/steps/slots.py +++ b/features/steps/slots.py @@ -15,17 +15,25 @@ def create_logical_replication_slot(context, slot_name, pg_name, plugin): assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin) -@then('{pg_name:w} has a logical replication slot named {slot_name} with the {plugin:w} plugin') -def has_logical_replication_slot(context, pg_name, slot_name, plugin): - try: - row = context.pctl.query(pg_name, ("SELECT slot_type, plugin FROM pg_replication_slots" - " WHERE slot_name = '{0}'").format(slot_name)).fetchone() - assert row, "Couldn't find replication slot named {0}".format(slot_name) - 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: - assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin) +@step('{pg_name:w} has a logical replication slot named {slot_name}' + ' with the {plugin:w} plugin after {time_limit:d} seconds') +@then('{pg_name:w} has a logical replication slot named {slot_name}' + ' with the {plugin:w} plugin after {time_limit:d} seconds') +def has_logical_replication_slot(context, pg_name, slot_name, plugin, time_limit): + time_limit *= context.timeout_multiplier + max_time = time.time() + int(time_limit) + while time.time() < max_time: + try: + row = context.pctl.query(pg_name, ("SELECT slot_type, plugin FROM pg_replication_slots" + f" WHERE slot_name = '{slot_name}'")).fetchone() + if row: + assert row[0] == "logical", f"Replication slot {slot_name} isn't a logical but {row[0]}" + assert row[1] == plugin, f"Replication slot {slot_name} using plugin {row[1]} rather than {plugin}" + return + except Exception: + pass + time.sleep(1) + assert False, f"Error looking for slot {slot_name} on {pg_name} with plugin {plugin}" @then('{pg_name:w} does not have a logical replication slot named {slot_name}') @@ -38,13 +46,14 @@ def does_not_have_logical_replication_slot(context, pg_name, slot_name): 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): +@step('{slot_type:w} slot {slot_name:w} is in sync between {pg_name1:w} and {pg_name2:w} after {time_limit:d} seconds') +def slots_in_sync(context, slot_type, slot_name, pg_name1, pg_name2, time_limit): time_limit *= context.timeout_multiplier max_time = time.time() + int(time_limit) + column = 'confirmed_flush_lsn' if slot_type.lower() == 'logical' else 'restart_lsn' + query = f"SELECT {column} FROM pg_replication_slots WHERE slot_name = '{slot_name}'" 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]: @@ -52,9 +61,31 @@ def logical_slots_in_sync(context, slot_name, pg_name1, pg_name2, time_limit): 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) + assert False, \ + f"{slot_type} slot {slot_name} is not in sync between {pg_name1} and {pg_name2} after {time_limit} seconds" @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)) + + +@step('I get all changes from physical slot {slot_name:w} on {pg_name:w}') +def physical_slot_get_changes(context, slot_name, pg_name): + context.pctl.query(pg_name, f"SELECT * FROM pg_replication_slot_advance('{slot_name}', pg_current_wal_lsn())") + + +@step('{pg_name:w} has a physical replication slot named {slot_name} after {time_limit:d} seconds') +def has_physical_replication_slot(context, pg_name, slot_name, time_limit): + time_limit *= context.timeout_multiplier + max_time = time.time() + int(time_limit) + query = f"SELECT * FROM pg_catalog.pg_replication_slots WHERE slot_type = 'physical' AND slot_name = '{slot_name}'" + while time.time() < max_time: + try: + row = context.pctl.query(pg_name, query).fetchone() + if row: + return + except Exception: + pass + time.sleep(1) + assert False, f"Physical slot {slot_name} doesn't exist after {time_limit} seconds" diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 87d3c9c1..08112651 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -28,6 +28,7 @@ from ..tags import Tags if TYPE_CHECKING: # pragma: no cover from ..config import Config +SLOT_ADVANCE_AVAILABLE_VERSION = 110000 CITUS_COORDINATOR_GROUP_ID = 0 citus_group_re = re.compile('^(0|[1-9][0-9]*)$') slot_name_re = re.compile('^[a-z0-9_]{1,63}$') @@ -350,6 +351,11 @@ class Member(Tags, NamedTuple('Member', logger.debug('Failed to parse Patroni version %s', version) return None + @property + def lsn(self) -> Optional[int]: + """Current LSN (receive/flush/replay).""" + return self.data.get('xlog_location') + class RemoteMember(Member): """Represents a remote member (typically a primary) for a standby cluster. @@ -964,24 +970,42 @@ class Cluster(NamedTuple('Cluster', candidates = [m for m in self.members if m.clonefrom and m.is_running and m.name not in exclude] return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader + @staticmethod + def is_physical_slot(value: Union[Any, Dict[str, Any]]) -> bool: + """Check whether provided configuration is for permanent physical replication slot. + + :returns: ``True`` if this is a physical replication slot, otherwise ``False``. + """ + return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical' + @property def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]: """Dictionary of permanent replication slots with their known LSN.""" - ret = deepcopy(self.config.permanent_slots if self.config else {}) - # If primary reported flush LSN for permanent slots we want to enrich our structure with it - for name, lsn in (self.slots or {}).items(): - if name in ret: - if not ret[name]: - ret[name] = {} - if isinstance(ret[name], dict): - ret[name]['lsn'] = lsn + leader = self.leader and self.leader.member + leader_name = slot_name_from_member_name(leader.name) if leader and leader.lsn else None + + slots = self.slots or {} + ret: Dict[str, Union[Dict[str, Any], Any]] = deepcopy(self.config.permanent_slots if self.config else {}) + + for name, value in list(ret.items()): + if not value: + value = ret[name] = {} + if isinstance(value, dict): + if name in slots: + # If primary reported flush LSN for permanent slots we want to enrich our structure with it + value['lsn'] = slots[name] + elif self.is_physical_slot(value) and name == leader_name and leader and leader.lsn: + # there is no slot on the leader for itself, use `lsn` from the member key. + value['lsn'] = leader.lsn + else: + # Don't let anyone set 'lsn' in the global configuration :) + value.pop('lsn', None) return ret @property def __permanent_physical_slots(self) -> Dict[str, Any]: """Dictionary of permanent ``physical`` replication slots.""" - return {name: value for name, value in self.__permanent_slots.items() - if not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'} + return {name: value for name, value in self.__permanent_slots.items() if self.is_physical_slot(value)} @property def __permanent_logical_slots(self) -> Dict[str, Any]: @@ -1013,7 +1037,7 @@ class Cluster(NamedTuple('Cluster', :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks. """ slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role) - permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover) + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover, major_version) disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( slots, permanent_slots, my_name, major_version) @@ -1061,7 +1085,7 @@ class Cluster(NamedTuple('Cluster', continue if value['type'] == 'logical' and value.get('database') and value.get('plugin'): - if major_version < 110000: + if major_version < SLOT_ADVANCE_AVAILABLE_VERSION: disabled_permanent_logical_slots.append(name) elif name in slots: logger.error("Permanent logical replication slot {'%s': %s} is conflicting with" @@ -1073,7 +1097,8 @@ class Cluster(NamedTuple('Cluster', logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name]) return disabled_permanent_logical_slots - def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]: + def _get_permanent_slots(self, is_standby_cluster: bool, role: str, + nofailover: bool, major_version: int) -> Dict[str, Any]: """Get configured permanent replication slots. .. note:: @@ -1089,6 +1114,7 @@ class Cluster(NamedTuple('Cluster', the outside because we want to protect from the ``/config`` key removal. :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``. :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param major_version: postgresql major version. :returns: dictionary of permanent slot names mapped to attributes. """ @@ -1096,9 +1122,11 @@ class Cluster(NamedTuple('Cluster', return {} if is_standby_cluster: - return self.__permanent_physical_slots if role == 'standby_leader' else {} + return self.__permanent_physical_slots \ + if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION or role == 'standby_leader' else {} - return self.__permanent_slots if role in ('master', 'primary') else self.__permanent_logical_slots + return self.__permanent_slots if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION\ + or role in ('master', 'primary') else self.__permanent_logical_slots def _get_members_slots(self, my_name: str, role: str) -> Dict[str, Dict[str, str]]: """Get physical replication slots configuration for members that sourcing from this node. @@ -1143,21 +1171,33 @@ class Cluster(NamedTuple('Cluster', for k, v in slot_conflicts.items() if len(v) > 1)) return slots - def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool: + def has_permanent_slots(self, my_name: str, nofailover: bool = False) -> bool: + """Check if the given member node has permanent replication slots configured. + + :param my_name: name of the member node to check. + :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + + :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``. + """ + members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, 'replica') + permanent_slots: Dict[str, Any] = self._get_permanent_slots(nofailover, 'replica', False, + SLOT_ADVANCE_AVAILABLE_VERSION) + slots = deepcopy(members_slots) + self._merge_permanent_slots(slots, permanent_slots, my_name, SLOT_ADVANCE_AVAILABLE_VERSION) + return len(slots) > len(members_slots) + + def _has_permanent_logical_slots(self, my_name: str, nofailover: bool) -> bool: """Check if the given member node has permanent ``logical`` replication slots configured. :param my_name: name of the member node to check. :param nofailover: ``True`` if this node is tagged to not be a failover candidate. - :param major_version: the PostgreSQL major version number. - :returns: ``False`` if PostgreSQL is < 11, ``True`` if any detected replications slots are ``logical``. + :returns: ``True`` if any detected replications slots are ``logical``, otherwise ``False``. """ - if major_version < 110000: - return False - slots = self.get_replication_slots(my_name, 'replica', nofailover, major_version).values() + slots = self.get_replication_slots(my_name, 'replica', nofailover, SLOT_ADVANCE_AVAILABLE_VERSION).values() return any(v for v in slots if v.get("type") == "logical") - def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool, major_version: int) -> bool: + def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool) -> bool: """Determine whether ``hot_standby_feedback`` should be enabled for the given member. The ``hot_standby_feedback`` must be enabled if the current replica has ``logical`` slots, @@ -1165,20 +1205,16 @@ class Cluster(NamedTuple('Cluster', :param my_name: name of the member node to check. :param nofailover: ``True`` if this node is tagged to not be a failover candidate. - :param major_version: PostgreSQL major version number. - :returns: ``True`` if this node or any member replicating from this node has permanent logical slots. - ``False`` if PostgreSQL major version is < 11. + :returns: ``True`` if this node or any member replicating from this node has + permanent logical slots, otherwise ``False``. """ - if major_version < 110000: - return False - - if self.has_permanent_logical_slots(my_name, nofailover, major_version): + if self._has_permanent_logical_slots(my_name, nofailover): 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 any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover) for m in members) return False def get_my_slot_name_on_primary(self, my_name: str, replicatefrom: Optional[str]) -> str: diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index e3cc1d69..6390aa2e 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -301,7 +301,7 @@ class ZooKeeper(AbstractDCS): raise ZooKeeperError('ZooKeeper in not responding properly') # The /status ZNode was updated or doesn't exist 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: + or cluster.has_permanent_slots(self._name) and not cluster.slots: # If current node is the leader just clear the event without fetching anything (we are updating the /status) if cluster.leader and cluster.leader.name == self._name: self.event.clear() diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 9bd53dac..af48886f 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -27,7 +27,7 @@ from .sync import SyncHandler from .. import psycopg from ..async_executor import CriticalTask from ..collections import CaseInsensitiveSet -from ..dcs import Cluster, Leader, Member +from ..dcs import Cluster, Leader, Member, SLOT_ADVANCE_AVAILABLE_VERSION from ..exceptions import PostgresConnectionException from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int @@ -112,7 +112,7 @@ class Postgresql(object): self._state_entry_timestamp = 0 self._cluster_info_state = {} - self._has_permanent_logical_slots = True + self._has_permanent_slots = True self._enforce_hot_standby_feedback = False self._cached_replica_timeline = None @@ -174,6 +174,11 @@ class Postgresql(object): """:returns: `True` if Postgres version supports more than one synchronous node.""" return self._major_version >= 90600 + @property + def can_advance_slots(self) -> bool: + """``True`` if :attr:``major_version`` is greater than 110000.""" + return self.major_version >= SLOT_ADVANCE_AVAILABLE_VERSION + @property def cluster_info_query(self) -> str: """Returns the monitoring query with a fixed number of fields. @@ -208,8 +213,9 @@ class Postgresql(object): extra = ("pg_catalog.current_setting('restore_command')" if self._major_version >= 120000 else "NULL") +\ ", " + ("(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 + " AS confirmed_flush_lsn, pg_catalog.pg_wal_lsn_diff(restart_lsn, '0/0')::bigint" + " AS restart_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)" + if self._has_permanent_slots and self.can_advance_slots else "NULL") + extra extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END," " slot_name, conninfo, status, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra) if self.role == 'standby_leader': @@ -434,18 +440,15 @@ class Postgresql(object): return if self._global_config.is_standby_cluster: + self._has_permanent_slots = False # Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback - self._has_permanent_logical_slots = False self.set_enforce_hot_standby_feedback(False) elif cluster and cluster.config and cluster.config.modify_version: - self._has_permanent_logical_slots =\ - cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version) - + self._has_permanent_slots = cluster.has_permanent_slots(self.name, nofailover) # We want to enable hot_standby_feedback if the replica is supposed # to have a logical slot or in case if it is the cascading replica. self.set_enforce_hot_standby_feedback( - self._has_permanent_logical_slots - or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) + self.can_advance_slots and cluster.should_enforce_hot_standby_feedback(self.name, nofailover)) def _cluster_info_state_get(self, name: str) -> Optional[Any]: if not self._cluster_info_state: @@ -456,7 +459,7 @@ class Postgresql(object): 'received_tli', 'slot_name', 'conninfo', 'receiver_state', 'restore_command', 'slots', 'synchronous_commit', 'synchronous_standby_names', 'pg_stat_replication'], result)) - if self._has_permanent_logical_slots: + if self._has_permanent_slots and self.can_advance_slots: cluster_info_state['slots'] =\ self.slots_handler.process_permanent_slots(cluster_info_state['slots']) self._cluster_info_state = cluster_info_state diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index f6090a3b..29bbb130 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -16,6 +16,7 @@ from .misc import format_lsn, fsync_dir from ..dcs import Cluster, Leader from ..file_perm import pg_perm from ..psycopg import OperationalError +from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from psycopg import Cursor @@ -231,15 +232,16 @@ class SlotsHandler: ret: Dict[str, int] = {} slots_dict: Dict[str, Dict[str, Any]] = {slot['slot_name']: slot for slot in slots or []} - if slots_dict: - for name, value in slots_dict.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]) + for name, value in slots_dict.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 + self._replication_slots[name]['restart_lsn'] = ret[name] = value['restart_lsn'] + else: + self._schedule_load_slots = True # It could happen that the slot was deleted in the background, we want to detect this case if any(name not in slots_dict for name in self._replication_slots.keys()): @@ -260,16 +262,19 @@ class SlotsHandler: """ if self._postgresql.major_version >= 90400 and self._schedule_load_slots: replication_slots: Dict[str, Dict[str, Any]] = {} - extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" \ + pg_wal_lsn_diff = f"pg_catalog.pg_{self._postgresql.wal_name}_{self._postgresql.lsn_name}_diff" + extra = f", catalog_xmin, {pg_wal_lsn_diff}(confirmed_flush_lsn, '0/0')::bigint" \ if self._postgresql.major_version >= 100000 else "" skip_temp_slots = ' WHERE NOT temporary' if self._postgresql.major_version >= 100000 else '' - for r in self._query('SELECT slot_name, slot_type, plugin, database, datoid' - f'{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}'): + for r in self._query(f"SELECT slot_name, slot_type, {pg_wal_lsn_diff}(restart_lsn, '0/0')::bigint, plugin," + f" database, datoid{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}"): value = {'type': r[1]} if r[1] == 'logical': - value.update(plugin=r[2], database=r[3], datoid=r[4]) + value.update(plugin=r[3], database=r[4], datoid=r[5]) if self._postgresql.major_version >= 100000: - value.update(catalog_xmin=r[5], confirmed_flush_lsn=r[6]) + value.update(catalog_xmin=r[6], confirmed_flush_lsn=r[7]) + else: + value['restart_lsn'] = r[2] replication_slots[r[0]] = value self._replication_slots = replication_slots self._schedule_load_slots = False @@ -353,7 +358,7 @@ class SlotsHandler: self._schedule_load_slots = True def _ensure_physical_slots(self, slots: Dict[str, Any]) -> None: - """Create any missing physical replication *slots*. + """Create or advance physical replication *slots*. Any failures are logged and do not interrupt creation of all *slots*. @@ -362,7 +367,9 @@ class SlotsHandler: """ 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': + if value['type'] != 'physical': + continue + if name not in self._replication_slots: try: self._query(f"SELECT pg_catalog.pg_create_physical_replication_slot(%s{immediately_reserve})" f" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" @@ -371,6 +378,16 @@ class SlotsHandler: except Exception: logger.exception("Failed to create physical replication slot '%s'", name) self._schedule_load_slots = True + elif not self._postgresql.is_primary() and self._postgresql.can_advance_slots \ + and self._replication_slots[name]['type'] == 'physical': + value['restart_lsn'] = self._replication_slots[name]['restart_lsn'] + lsn = parse_int(value.get('lsn')) + if lsn and lsn > value['restart_lsn']: # The slot has feedback in DCS and needs to be advanced + try: + lsn = format_lsn(lsn) + self._query("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", name, lsn) + except Exception as exc: + logger.error("Error while advancing replication slot %s to position '%s': %r", name, lsn, exc) @contextmanager def get_local_connection_cursor(self, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]: @@ -484,10 +501,11 @@ class SlotsHandler: replicatefrom: Optional[str] = None, paused: bool = False) -> List[str]: """During the HA loop read, check and alter replication slots found in the cluster. - Read physical and logical slots found on the primary, then compare to those configured in the DCS. + Read physical and logical slots from ``pg_replication_slots``, then compare to those configured in the DCS. Drop any slots that do not match those required by configuration and are not configured as permanent. - Create any missing physical slots. If we are the leader then logical slots too, otherwise if logical slots - are known and active create them on replica nodes. + Create any missing physical slots, or advance their position according to feedback stored in DCS. + If we are the primary then create logical slots, otherwise if logical slots are known and active create + them on replica nodes by copying slot files from the primary. :param cluster: object containing stateful information for the cluster. :param nofailover: ``True`` if this node has been tagged to not be a failover candidate. diff --git a/tests/__init__.py b/tests/__init__.py index 68961af7..5240cec6 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -104,12 +104,14 @@ class MockCursor(object): elif sql.startswith('SELECT slot_name, slot_type, datname, plugin, catalog_xmin'): self.results = [('ls', 'logical', 'a', 'b', 100, 500, b'123456')] elif sql.startswith('SELECT slot_name'): - self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'b', 'a', 5, 100, 500)] + self.results = [('blabla', 'physical', 12345), + ('foobar', 'physical', 12345), + ('ls', 'logical', 499, 'b', 'a', 5, 100, 500)] elif sql.startswith('WITH slots AS (SELECT slot_name, active'): self.results = [(False, True)] if self.rowcount == 1 else [] elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'): self.results = [(1, 2, 1, 0, False, 1, 1, None, None, 'streaming', '', - [{"slot_name": "ls", "confirmed_flush_lsn": 12345}], + [{"slot_name": "ls", "confirmed_flush_lsn": 12345, "restart_lsn": 12344}], 'on', 'n1', None)] elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'): self.results = [(False, 2)] @@ -252,8 +254,8 @@ class BaseTestPostgresql(PostgresInit): if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) - self.leadermem = Member(0, 'leader', 28, { - 'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'}) + self.leadermem = Member(0, 'leader', 28, {'xlog_location': 100, 'state': 'running', + 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'}) self.leader = Leader(-1, 28, self.leadermem) self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', 'state': 'running', 'tags': {'replicatefrom': 'leader'}}) diff --git a/tests/test_slots.py b/tests/test_slots.py index 884f76b1..d1b8d458 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -53,6 +53,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.p.set_role('replica') with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \ patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop: + config.data['slots'].pop('ls') self.s.sync_replication_slots(cluster, False, paused=True) mock_drop.assert_not_called() self.p.set_role('primary') @@ -69,6 +70,8 @@ class TestSlotsHandler(BaseTestPostgresql): 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) + self.p.set_role('replica') + self.s.sync_replication_slots(cluster, False) def test_cascading_replica_sync_replication_slots(self): """Test sync with a cascading replica so physical slots are present on a replica.""" @@ -82,12 +85,12 @@ class TestSlotsHandler(BaseTestPostgresql): self.p.set_role('replica') with patch.object(Postgresql, '_query') as mock_query, \ patch.object(Postgresql, 'is_primary', Mock(return_value=False)): - mock_query.return_value = [('ls', 'logical', 'b', 'a', 5, 12345, 105)] + mock_query.return_value = [('ls', 'logical', 104, 'b', 'a', 5, 12345, 105)] ret = self.s.sync_replication_slots(cluster, False) self.assertEqual(ret, []) def test_process_permanent_slots(self): - config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}, + config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}, 'blabla': {'type': 'physical'}}, 'ignore_slots': [{'name': 'blabla'}]}, 1) cluster = Cluster(True, config, self.leader, Status.empty(), [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None) @@ -98,8 +101,10 @@ class TestSlotsHandler(BaseTestPostgresql): mock_query.return_value = [( 1, 0, 0, 0, 0, 0, 0, 0, 0, None, None, [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b", - "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])] - self.assertEqual(self.p.slots(), {'ls': 12345}) + "confirmed_flush_lsn": 12345, "catalog_xmin": 105, "restart_lsn": 12344}, + {"slot_name": "blabla", "type": "physical", "datoid": None, "plugin": None, + "confirmed_flush_lsn": None, "catalog_xmin": 105, "restart_lsn": 12344}])] + self.assertEqual(self.p.slots(), {'ls': 12345, 'blabla': 12344}) self.p.reset_cluster_info_state(None) mock_query.return_value = [( @@ -114,8 +119,8 @@ class TestSlotsHandler(BaseTestPostgresql): self.cluster.slots['ls'] = 12346 with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) - self.s._schedule_load_slots = False - with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \ + with patch.object(SlotsHandler, '_query', Mock(return_value=[('ls', 'logical', 499, 'b', 'a', 5, 100, 500)])), \ + patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \ patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \ patch.object(psycopg.OperationalError, 'diag') as mock_diag: type(mock_diag).sqlstate = PropertyMock(return_value='58P01') @@ -177,3 +182,18 @@ class TestSlotsHandler(BaseTestPostgresql): with patch.object(SlotsHandler, 'get_local_connection_cursor', Mock(side_effect=Exception)): self.s.schedule_advance_slots({'foo': {'bar': 100}}) self.s._advance.sync_slots() + + @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) + def test_advance_physical_slots(self): + config = ClusterConfig(1, {'slots': {'blabla': {'type': 'physical'}, 'leader': None}}, 1) + cluster = Cluster(True, config, self.leader, Status(0, {'blabla': 12346}), + [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None) + self.s.sync_replication_slots(cluster, False) + with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('blabla', 'physical', 12345, None, None, None, + None, None)], Exception])) as mock_query, \ + patch('patroni.postgresql.slots.logger.error') as mock_error: + self.s.sync_replication_slots(cluster, False) + self.assertEqual(mock_query.call_args[0], + ("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", "blabla", '0/303A')) + self.assertEqual(mock_error.call_args[0][0], + "Error while advancing replication slot %s to position '%s': %r")