From 704d36815aa418438eb73c87be0c3771246d0c84 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 17 Aug 2023 12:33:15 +0200 Subject: [PATCH 1/6] Explicitly enable synchronous mode (#2820) Close https://github.com/zalando/patroni/issues/2819 Co-authored-by: Polina Bungina <27892524+hughcapet@users.noreply.github.com> --- patroni/ha.py | 13 +++++++++++-- tests/test_ha.py | 13 +++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index f91e93be..28bd92a3 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -656,11 +656,20 @@ class Ha(object): promoting standbys that were guaranteed to be replicating synchronously. """ if self.is_synchronous_mode(): - current = CaseInsensitiveSet(self.cluster.sync.members) + sync = self.cluster.sync + if sync.is_empty: + # corner case: we need to explicitly enable synchronous mode by updating the + # ``/sync`` key with the current leader name and empty members. In opposite case + # it will never be automatically enabled if there are not eligible candidates. + sync = self.dcs.write_sync_state(self.state_handler.name, None, version=sync.version) + if not sync: + return logger.warning("Updating sync state failed") + logger.info("Enabled synchronous replication") + + current = CaseInsensitiveSet(sync.members) picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster) if picked != current: - sync = self.cluster.sync # update synchronous standby list in dcs temporarily to point to common nodes in current and picked sync_common = current & allow_promote if sync_common != current: diff --git a/tests/test_ha.py b/tests/test_ha.py index f7378b43..6acb2e15 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1294,6 +1294,19 @@ class TestHa(PostgresInit): mock_restart.assert_called_once() self.ha.dcs.get_cluster.assert_not_called() + def test_enable_synchronous_mode(self): + self.ha.is_synchronous_mode = true + self.ha.has_lock = true + self.p.name = 'leader' + self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) + with patch('patroni.ha.logger.info') as mock_logger: + self.ha.run_cycle() + self.assertEqual(mock_logger.call_args[0][0], 'Enabled synchronous replication') + self.ha.dcs.write_sync_state = Mock(return_value=None) + with patch('patroni.ha.logger.warning') as mock_logger: + self.ha.run_cycle() + self.assertEqual(mock_logger.call_args[0][0], 'Updating sync state failed') + def test_effective_tags(self): self.ha._disable_sync = True self.assertEqual(self.ha.get_effective_tags(), {'foo': 'bar', 'nosync': True}) From a4ac4963d1bcaac3b7326c65f191d3fa8341dc61 Mon Sep 17 00:00:00 2001 From: Israel Date: Thu, 17 Aug 2023 07:55:42 -0300 Subject: [PATCH 2/6] Fix `IntValidator` regarding validation of value `0` (#2818) Previous to this commit `IntValidator` would always consider the value `0` invalid, even if in the allowed range. The problem was that `parse_int` was returning `0` in the following line: ```python value = parse_int(value, self.base_unit) or "" ``` However the `or ""` was evaluating to an empty string. As `parse_int` returns either an `int` if able to parse, or `None` otherwise, the `isinstance(value, int)` is enough to error out when not a valid `int`. Closes #2817 --- patroni/validator.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/validator.py b/patroni/validator.py index c99b3d32..ddfb2c07 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -799,7 +799,7 @@ class IntValidator(object): :returns: ``True`` if *value* is valid and within the expected range. """ - value = parse_int(value, self.base_unit) or "" + value = parse_int(value, self.base_unit) ret = isinstance(value, int)\ and (self.min is None or value >= self.min)\ and (self.max is None or value <= self.max) From 899cad1c0f7cf2630e5d6c2e9631e7d531348109 Mon Sep 17 00:00:00 2001 From: Jelte Fennema Date: Thu, 17 Aug 2023 13:18:37 +0200 Subject: [PATCH 3/6] Remove Python 2 install instructions from README (#2821) --- README.rst | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/README.rst b/README.rst index 84e67379..104f0907 100644 --- a/README.rst +++ b/README.rst @@ -74,9 +74,8 @@ There are a few options available: :: - sudo apt-get install python-psycopg2 # install python2 psycopg2 module on Debian/Ubuntu - sudo apt-get install python3-psycopg2 # install python3 psycopg2 module on Debian/Ubuntu - sudo yum install python-psycopg2 # install python2 psycopg2 on RedHat/Fedora/CentOS + sudo apt-get install python3-psycopg2 # install psycopg2 module on Debian/Ubuntu + sudo yum install python3-psycopg2 # install psycopg2 on RedHat/Fedora/CentOS 2. Install psycopg2 from the binary package From 366829e3791097a233aa194c51b5f149d0163107 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 17 Aug 2023 15:42:11 +0200 Subject: [PATCH 4/6] Refactor Connection class (#2815) 1. stop using the same cursor all the time, it creates problems when not carefully used from different threads. 2. introduce query() method in the Connection class and make it return a result set when it is possible. 3. refactor most of the code that is relying (directly or indirectly) on the Connection object to use the query() method as much as possible. This refactoring helps with reducing code complexity and will help with future introduction of a separate database connection for the REST API thread. The last one will help to improve reliability when system is under significant stress when simple monitoring queries are taking seconds to execute and the REST API starts blocking the main thread. --- patroni/api.py | 25 +++----- patroni/postgresql/__init__.py | 101 ++++++++++++++++--------------- patroni/postgresql/citus.py | 20 +++--- patroni/postgresql/config.py | 4 +- patroni/postgresql/connection.py | 63 +++++++++++++++---- patroni/postgresql/slots.py | 31 ++++------ tests/__init__.py | 10 +-- tests/test_api.py | 19 +++--- tests/test_postgresql.py | 15 +++-- tests/test_rewind.py | 5 +- tests/test_slots.py | 12 ++-- 11 files changed, 165 insertions(+), 140 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 18170334..b01c24d8 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -1183,20 +1183,18 @@ class RestApiHandler(BaseHTTPRequestHandler): self.command = mname return ret - def query(self, sql: str, *params: Any, **kwargs: Any) -> List[Tuple[Any, ...]]: - """Execute *sql* query with *params*. + def query(self, sql: str, *params: Any, retry: bool = False) -> List[Tuple[Any, ...]]: + """Execute *sql* query with *params* and optionally return results. :param sql: the SQL statement to be run. :param params: positional arguments to call :func:`RestApiServer.query` with. - :param kwargs: can contain the key ``retry``. If the key is present its value should be a :class:`bool` which - indicates whether the query should be retried upon failure or given up immediately. + :param retry: whether the query should be retried upon failure or given up immediately. :returns: a list of rows that were fetched from the database. """ - if not kwargs.get('retry', False): + if not retry: return self.server.query(sql, *params) - retry = Retry(delay=1, retry_exceptions=PostgresConnectionException) - return retry(self.server.query, sql, *params) + return Retry(delay=1, retry_exceptions=PostgresConnectionException)(self.server.query, sql, *params) def get_postgresql_status(self, retry: bool = False) -> Dict[str, Any]: """Builds an object representing a status of "postgres". @@ -1368,7 +1366,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): self.daemon = True def query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: - """Execute *sql* query with *params*. + """Execute *sql* query with *params* and optionally return results. :param sql: the SQL statement to be run. :param params: positional arguments to be used as parameters for *sql*. @@ -1379,15 +1377,10 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): :class:`psycopg.Error`: if had issues while executing *sql*. :class:`~patroni.exceptions.PostgresConnectionException`: if had issues while connecting to the database. """ - cursor = None try: - with self.patroni.postgresql.connection().cursor() as cursor: - cursor.execute(sql.encode('utf-8'), params) - return [r for r in cursor] - except psycopg.Error as e: - if cursor and cursor.connection.closed == 0: - raise e - raise PostgresConnectionException('connection problems') + return self.patroni.postgresql.query(sql, *params, retry=False) + except RetryFailedError as e: + raise PostgresConnectionException(str(e)) @staticmethod def _set_fd_cloexec(fd: socket.socket) -> None: diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index af87c3e2..7f3bdccb 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -332,36 +332,50 @@ class Postgresql(object): self._connection.set_conn_kwargs(kwargs.copy()) self.citus_handler.set_conn_kwargs(kwargs.copy()) - def _query(self, sql: str, *params: Any) -> Union['Cursor[Any]', 'cursor']: - """We are always using the same cursor, therefore this method is not thread-safe!!! - You can call it from different threads only if you are holding explicit `AsyncExecutor` lock, - because the main thread is always holding this lock when running HA cycle.""" - cursor = None - try: - cursor = self._connection.cursor() - cursor.execute(sql.encode('utf-8'), params or None) - return cursor - except psycopg.Error as e: - if cursor and cursor.connection.closed == 0: - # When connected via unix socket, psycopg2 can't recoginze 'connection lost' - # and leaves `_cursor_holder.connection.closed == 0`, but psycopg2.OperationalError - # is still raised (what is correct). It doesn't make sense to continiue with existing - # connection and we will close it, to avoid its reuse by the `cursor` method. - if isinstance(e, psycopg.OperationalError): - self._connection.close() - else: - raise e - if self.state == 'restarting': - raise RetryFailedError('cluster is being restarted') - raise PostgresConnectionException('connection problems') + def _query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: + """Execute *sql* query with *params* and optionally return results. - def query(self, sql: str, *args: Any, **kwargs: Any) -> Union['Cursor[Any]', 'cursor']: - if not kwargs.get('retry', True): - return self._query(sql, *args) + :param sql: SQL statement to execute. + :param params: parameters to pass. + + :returns: a query response as a list of tuples if there is any. + :raises: + :exc:`~psycopg.Error` if had issues while executing *sql*. + + :exc:`~patroni.exceptions.PostgresConnectionException`: if had issues while connecting to the database. + + :exc:`~patroni.utils.RetryFailedError`: if it was detected that connection/query failed due to PostgreSQL + restart. + """ try: - return self.retry(self._query, sql, *args) - except RetryFailedError as e: - raise PostgresConnectionException(str(e)) + return self._connection.query(sql, *params) + except PostgresConnectionException as exc: + if self.state == 'restarting': + raise RetryFailedError('cluster is being restarted') from exc + raise + + def query(self, sql: str, *params: Any, retry: bool = True) -> List[Tuple[Any, ...]]: + """Execute *sql* query with *params* and optionally return results. + + :param sql: SQL statement to execute. + :param params: parameters to pass. + :param retry: whether the query should be retried upon failure or given up immediately. + + :returns: a query response as a list of tuples if there is any. + :raises: + :exc:`~psycopg.Error` if had issues while executing *sql*. + + :exc:`~patroni.exceptions.PostgresConnectionException`: if had issues while connecting to the database. + + :exc:`~patroni.utils.RetryFailedError`: if it was detected that connection/query failed due to PostgreSQL + restart or if retry deadline was exceeded. + """ + if not retry: + return self._query(sql, *params) + try: + return self.retry(self._query, sql, *params) + except RetryFailedError as exc: + raise PostgresConnectionException(str(exc)) from exc def pg_control_exists(self) -> bool: return os.path.isfile(self._pg_control) @@ -431,7 +445,7 @@ class Postgresql(object): def _cluster_info_state_get(self, name: str) -> Optional[Any]: if not self._cluster_info_state: try: - result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone() + result = self._is_leader_retry(self._query, self.cluster_info_query)[0] cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location', 'received_location', 'replay_paused', 'pg_control_timeline', 'received_tli', 'slot_name', 'conninfo', 'receiver_state', @@ -873,11 +887,10 @@ class Postgresql(object): def _wait_for_connection_close(self, postmaster: PostmasterProcess) -> None: try: - with self.connection().cursor() as cur: - while postmaster.is_running(): # Need a timeout here? - cur.execute("SELECT 1") - time.sleep(STOP_POLLING_INTERVAL) - except psycopg.Error: + while postmaster.is_running(): # Need a timeout here? + self._connection.query("SELECT 1") + time.sleep(STOP_POLLING_INTERVAL) + except (psycopg.Error, PostgresConnectionException): pass def reload(self, block_callbacks: bool = False) -> bool: @@ -1178,26 +1191,16 @@ class Postgresql(object): received_location = self.received_location() pg_control_timeline = self._cluster_info_state_get('pg_control_timeline') else: - with self.connection().cursor() as cursor: - cursor.execute(self.cluster_info_query.encode('utf-8')) - row = cursor.fetchone() - if TYPE_CHECKING: # pragma: no cover - assert row is not None - (timeline, wal_position, replayed_location, received_location, _, pg_control_timeline) = row[:6] + timeline, wal_position, replayed_location, received_location, _, pg_control_timeline = \ + self._query(self.cluster_info_query)[0][:6] wal_position = self._wal_position(bool(timeline), wal_position, received_location, replayed_location) - return (timeline, wal_position, pg_control_timeline) + return timeline, wal_position, pg_control_timeline def postmaster_start_time(self) -> Optional[str]: try: - query = "SELECT " + self.POSTMASTER_START_TIME - if current_thread().ident == self.__thread_ident: - row = self.query(query).fetchone() - else: - with self.connection().cursor() as cursor: - cursor.execute(query) - row = cursor.fetchone() - return row[0].isoformat(sep=' ') if row else None + sql = "SELECT " + self.POSTMASTER_START_TIME + return self.query(sql, retry=current_thread().ident == self.__thread_ident)[0][0].isoformat(sep=' ') except psycopg.Error: return None diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index a3a8fe45..f659e325 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -11,8 +11,6 @@ from ..dcs import CITUS_COORDINATOR_GROUP_ID, Cluster from ..psycopg import connect, quote_ident if TYPE_CHECKING: # pragma: no cover - from psycopg import Cursor - from psycopg2 import cursor from . import Postgresql CITUS_SLOT_NAME_RE = re.compile(r'^citus_shard_(move|split)_slot(_[1-9][0-9]*){2,3}$') @@ -109,12 +107,10 @@ class CitusHandler(Thread): self._tasks[:] = [] self._in_flight = None - def query(self, sql: str, *params: Any) -> Union['Cursor[Any]', 'cursor']: + def query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: try: logger.debug('query(%s, %s)', sql, params) - cursor = self._connection.cursor() - cursor.execute(sql.encode('utf-8'), params or None) - return cursor + return self._connection.query(sql, *params) except Exception as e: logger.error('Exception when executing query "%s", (%s): %r', sql, params, e) self._connection.close() @@ -132,13 +128,13 @@ class CitusHandler(Thread): self._schedule_load_pg_dist_node = False try: - cursor = self.query("SELECT nodeid, groupid, nodename, nodeport, noderole" - " FROM pg_catalog.pg_dist_node WHERE noderole = 'primary'") + rows = self.query("SELECT nodeid, groupid, nodename, nodeport, noderole" + " FROM pg_catalog.pg_dist_node WHERE noderole = 'primary'") except Exception: return False with self._condition: - self._pg_dist_node = {r[1]: PgDistNode(r[1], r[2], r[3], 'after_promote', r[0]) for r in cursor} + self._pg_dist_node = {r[1]: PgDistNode(r[1], r[2], r[3], 'after_promote', r[0]) for r in rows} return True def sync_pg_dist_node(self, cluster: Cluster) -> None: @@ -211,10 +207,8 @@ class CitusHandler(Thread): self.query('SELECT pg_catalog.citus_update_node(%s, %s, %s, true, %s)', task.nodeid, task.host, task.port, task.cooldown) elif task.event != 'before_demote': - row = self.query("SELECT pg_catalog.citus_add_node(%s, %s, %s, 'primary', 'default')", - task.host, task.port, task.group).fetchone() - if row is not None: - task.nodeid = row[0] + task.nodeid = self.query("SELECT pg_catalog.citus_add_node(%s, %s, %s, 'primary', 'default')", + task.host, task.port, task.group)[0][0] def process_task(self, task: PgDistNode) -> bool: """Updates a single row in `pg_dist_node` table, optionally in a transaction. diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 6d1fe384..51bbf9af 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -1093,10 +1093,10 @@ class ConfigHandler(object): if self._postgresql.major_version >= 90500: time.sleep(1) try: - pending_restart = (self._postgresql.query( + pending_restart = self._postgresql.query( 'SELECT COUNT(*) FROM pg_catalog.pg_settings' ' WHERE pg_catalog.lower(name) != ALL(%s) AND pending_restart', - [n.lower() for n in self._RECOVERY_PARAMETERS]).fetchone() or (0,))[0] > 0 + [n.lower() for n in self._RECOVERY_PARAMETERS])[0][0] > 0 self._postgresql.set_pending_restart(pending_restart) except Exception as e: logger.warning('Exception %r when running query', e) diff --git a/patroni/postgresql/connection.py b/patroni/postgresql/connection.py index 277a4889..5bb97a1f 100644 --- a/patroni/postgresql/connection.py +++ b/patroni/postgresql/connection.py @@ -2,45 +2,86 @@ import logging from contextlib import contextmanager from threading import Lock -from typing import Any, Dict, Iterator, Union, TYPE_CHECKING +from typing import Any, Dict, Iterator, List, Union, Tuple, TYPE_CHECKING if TYPE_CHECKING: # pragma: no cover from psycopg import Connection as Connection3, Cursor from psycopg2 import connection, cursor from .. import psycopg +from ..exceptions import PostgresConnectionException logger = logging.getLogger(__name__) -class Connection(object): +class Connection: + """Helper class to manage connections from Patroni to PostgreSQL. + + :ivar server_version: PostgreSQL version in integer format where we are connected to. + """ + server_version: int def __init__(self) -> None: - self._lock = Lock() + """Create an instance of :class:`Connection` class.""" + self._lock = Lock() # used to make sure that only one connection to postgres is established self._connection = None - self._cursor_holder = None def set_conn_kwargs(self, conn_kwargs: Dict[str, Any]) -> None: + """Set connection parameters, like user, password, host, port and so on. + + :param conn_kwargs: connection parameters as a dictionary. + """ self._conn_kwargs = conn_kwargs def get(self) -> Union['connection', 'Connection3[Any]']: + """Get ``psycopg``/``psycopg2`` connection object. + + .. note:: + Opens a new connection if necessary. + + :returns: ``psycopg`` or ``psycopg2`` connection object. + """ with self._lock: if not self._connection or self._connection.closed != 0: + logger.info("establishing a new patroni connection to postgres") self._connection = psycopg.connect(**self._conn_kwargs) self.server_version = getattr(self._connection, 'server_version', 0) return self._connection - def cursor(self) -> Union['cursor', 'Cursor[Any]']: - if not self._cursor_holder or self._cursor_holder.closed or self._cursor_holder.connection.closed != 0: - logger.info("establishing a new patroni connection to the postgres cluster") - self._cursor_holder = self.get().cursor() - return self._cursor_holder + def query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: + """Execute a query with parameters and optionally returns a response. + + :param sql: SQL statement to execute. + :param params: parameters to pass. + + :returns: a query response as a list of tuples if there is any. + :raises: + :exc:`~psycopg.Error` if had issues while executing *sql*. + + :exc:`~patroni.exceptions.PostgresConnectionException`: if had issues while connecting to the database. + """ + cursor = None + try: + with self.get().cursor() as cursor: + cursor.execute(sql.encode('utf-8'), params or None) + return cursor.fetchall() if cursor.rowcount and cursor.rowcount > 0 else [] + except psycopg.Error as exc: + if cursor and cursor.connection.closed == 0: + # When connected via unix socket, psycopg2 can't recoginze 'connection lost' and leaves + # `self._connection.closed == 0`, but the generic exception is raised. It doesn't make + # sense to continue with existing connection and we will close it, to avoid its reuse. + if type(exc) in (psycopg.DatabaseError, psycopg.OperationalError): + self.close() + else: + raise exc + raise PostgresConnectionException('connection problems') from exc def close(self) -> None: + """Close the psycopg connection to postgres.""" if self._connection and self._connection.closed == 0: self._connection.close() - logger.info("closed patroni connection to the postgresql cluster") - self._cursor_holder = self._connection = None + logger.info("closed patroni connection to postgres") + self._connection = None @contextmanager diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 681aed00..319119ce 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -186,7 +186,7 @@ class SlotsHandler: self.pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot') self.schedule() - def _query(self, sql: str, *params: Any) -> Union['cursor', 'Cursor[Any]']: + def _query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: """Helper method for :meth:`Postgresql.query`. :param sql: SQL statement to execute. @@ -263,9 +263,8 @@ class SlotsHandler: extra = ", catalog_xmin, pg_catalog.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 '' - cursor = self._query(f'SELECT slot_name, slot_type, plugin, database, datoid' - f'{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}') - for r in cursor: + for r in self._query('SELECT slot_name, slot_type, plugin, database, datoid' + f'{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]) @@ -308,16 +307,13 @@ class SlotsHandler: ``dropped`` is ``True`` if the slot was successfully dropped. If the slot was not found return ``False`` for both. """ - cursor = self._query(('WITH slots AS (SELECT slot_name, active' - ' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),' - ' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),' - ' true AS dropped FROM slots WHERE not active) ' - 'SELECT active, COALESCE(dropped, false) FROM slots' - ' FULL OUTER JOIN dropped ON true'), name) - row = cursor.fetchone() - if not row: - row = (False, False) - return row + rows = self._query(('WITH slots AS (SELECT slot_name, active' + ' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),' + ' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),' + ' 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) 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. @@ -595,11 +591,8 @@ class SlotsHandler: # Replica isn't streaming or the hot_standby_feedback isn't enabled try: - cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean") - row = cur.fetchone() - if row and not row[0]: - logger.error('Logical slot failover requires "hot_standby_feedback".' - ' Please check postgresql.auto.conf') + if not self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean")[0][0]: + logger.error('Logical slot failover requires "hot_standby_feedback". Please check postgresql.auto.conf') except Exception as e: logger.error('Failed to check the hot_standby_feedback setting: %r', e) return False diff --git a/tests/__init__.py b/tests/__init__.py index 01d74616..79ef0229 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -106,7 +106,7 @@ class MockCursor(object): elif sql.startswith('SELECT slot_name'): self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)] elif sql.startswith('WITH slots AS (SELECT slot_name, active'): - self.results = [(False, True)] if self.rowcount == 1 else [None] + 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}], @@ -114,10 +114,7 @@ class MockCursor(object): elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'): self.results = [(False, 2)] elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'): - replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\ - '"state":"streaming","sync_state":"async","sync_priority":0}]' - now = datetime.datetime.now(tzutc) - self.results = [(now, 0, '', 0, '', False, now, 'streaming', None, replication_info)] + self.results = [(datetime.datetime.now(tzutc),)] elif sql.startswith('SELECT name, current_setting(name) FROM pg_settings'): self.results = [('data_directory', 'data'), ('hba_file', os.path.join('data', 'pg_hba.conf')), @@ -143,6 +140,8 @@ class MockCursor(object): ('listen_addresses', '*', None, 'string', 'postmaster'), ('autovacuum', 'on', None, 'bool', 'sighup'), ('unix_socket_directories', '/tmp', None, 'string', 'postmaster')] + elif sql.startswith('SELECT COUNT(*) FROM pg_catalog.pg_settings'): + self.results = [(1,)] elif sql.startswith('IDENTIFY_SYSTEM'): self.results = [('1', 3, '0/402EEC0', '')] elif sql.startswith('TIMELINE_HISTORY '): @@ -156,6 +155,7 @@ class MockCursor(object): self.results = [(1, 0, 'host1', 5432, 'primary'), (2, 1, 'host2', 5432, 'primary')] else: self.results = [(None, None, None, None, None, None, None, None, None, None)] + self.rowcount = len(self.results) def fetchone(self): return self.results[0] diff --git a/tests/test_api.py b/tests/test_api.py index b5aeff8a..b1e56fdf 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -3,8 +3,6 @@ import json import unittest import socket -import patroni.psycopg as psycopg - from http.server import HTTPServer from io import BytesIO as IO from mock import Mock, PropertyMock, patch @@ -14,9 +12,8 @@ from patroni.api import RestApiHandler, RestApiServer from patroni.config import GlobalConfig from patroni.dcs import ClusterConfig, Member from patroni.ha import _MemberStatus -from patroni.utils import tzutc +from patroni.utils import RetryFailedError, tzutc -from . import psycopg_connect, MockCursor from .test_ha import get_cluster_initialized_without_leader @@ -41,10 +38,6 @@ class MockPostgresql(object): TL_LSN = 'CASE WHEN pg_catalog.pg_is_in_recovery()' citus_handler = Mock() - @staticmethod - def connection(): - return psycopg_connect() - @staticmethod def postmaster_start_time(): return postmaster_start_time @@ -61,6 +54,12 @@ class MockPostgresql(object): def replication_state_from_parameters(*args): return 'streaming' + @staticmethod + def query(sql, *params, retry=False): + return [(postmaster_start_time, 0, '', 0, '', False, postmaster_start_time, 'streaming', None, + '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' + + '"state":"streaming","sync_state":"async","sync_priority":0}]')] + class MockWatchdog(object): is_healthy = False @@ -488,9 +487,7 @@ class TestRestApiHandler(unittest.TestCase): @patch('time.sleep', Mock()) def test_RestApiServer_query(self): - with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)): - self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) - with patch.object(MockPostgresql, 'connection', Mock(side_effect=psycopg.OperationalError)): + with patch.object(MockPostgresql, 'query', Mock(side_effect=RetryFailedError('bla'))): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) @patch('time.sleep', Mock()) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 4f1e36da..510c2a90 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -545,9 +545,7 @@ class TestPostgresql(BaseTestPostgresql): @patch('time.sleep', Mock()) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) - @patch.object(MockCursor, 'fetchone') - def test_reload_config(self, mock_fetchone): - mock_fetchone.return_value = (1,) + def test_reload_config(self): parameters = self._PARAMETERS.copy() parameters.pop('f.oo') parameters['wal_buffers'] = '512' @@ -555,9 +553,14 @@ class TestPostgresql(BaseTestPostgresql): 'authentication': {}, 'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters} self.p.reload_config(config) - mock_fetchone.side_effect = Exception parameters['b.ar'] = 'bar' - self.p.reload_config(config) + with patch.object(MockCursor, 'fetchall', + Mock(side_effect=[[('wal_block_size', '8191', None, 'integer', 'internal'), + ('wal_segment_size', '2048', '8kB', 'integer', 'internal'), + ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'), + ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'), + ('port', '5433', None, 'integer', 'postmaster')], Exception])): + self.p.reload_config(config) parameters['autovacuum'] = 'on' self.p.reload_config(config) parameters['autovacuum'] = 'off' @@ -585,7 +588,7 @@ class TestPostgresql(BaseTestPostgresql): def test_postmaster_start_time(self): now = datetime.datetime.now() - with patch.object(MockCursor, "fetchone", Mock(return_value=(now, True, '', '', '', '', False))): + with patch.object(MockCursor, "fetchall", Mock(return_value=[(now, True, '', '', '', '', False)])): self.assertEqual(self.p.postmaster_start_time(), now.isoformat(sep=' ')) t = Thread(target=self.p.postmaster_start_time) t.start() diff --git a/tests/test_rewind.py b/tests/test_rewind.py index f40c46af..a54c27e9 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -92,8 +92,9 @@ class TestRewind(BaseTestPostgresql): self.r.rewind_or_reinitialize_needed_and_possible(self.leader) with patch.object(Postgresql, 'is_running', Mock(return_value=True)), \ - patch.object(MockCursor, 'fetchone', - Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])): + patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)), \ + patch.object(MockCursor, 'fetchall', + Mock(return_value=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None)])): 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 index 0210d701..df8c6955 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -93,17 +93,17 @@ class TestSlotsHandler(BaseTestPostgresql): 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 = ( + 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}]) + "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 = ( + mock_query.return_value = [( 1, 0, 0, 0, 0, 0, 0, 0, 0, None, None, [{"slot_name": "ls", "type": "logical", "datoid": 6, "plugin": "b", - "confirmed_flush_lsn": 12345, "catalog_xmin": 105}]) + "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])] self.assertEqual(self.p.slots(), {}) @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) @@ -137,10 +137,10 @@ class TestSlotsHandler(BaseTestPostgresql): def test_check_logical_slots_readiness(self): self.s.copy_logical_slots(self.cluster, ['ls']) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \ - patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)): + patch.object(MockCursor, 'fetchall', Mock(side_effect=Exception)): self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \ - patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))): + patch.object(MockCursor, 'fetchall', Mock(return_value=[(False,)])): self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))): self.s.check_logical_slots_readiness(self.cluster, None) From 93be10a6551a1767e94973be97d14ad2dfc9c359 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 17 Aug 2023 16:17:34 +0200 Subject: [PATCH 5/6] Remove Python 2 install instructions from docs/README (#2822) docs/README.rst mainly duplicates README.rst and also should be changed. Besides that remove test/coverage badges. followup on #2821 --- README.rst | 2 +- docs/README.rst | 9 ++------- 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/README.rst b/README.rst index 104f0907..f8f187e4 100644 --- a/README.rst +++ b/README.rst @@ -93,7 +93,7 @@ There are a few options available: :: - pip install psycopg[binary] + pip install psycopg[binary]>=3.0.0 **General installation for pip** diff --git a/docs/README.rst b/docs/README.rst index 4e3e7529..daba743d 100644 --- a/docs/README.rst +++ b/docs/README.rst @@ -46,9 +46,8 @@ There are a few options available: :: - sudo apt-get install python-psycopg2 # install python2 psycopg2 module on Debian/Ubuntu - sudo apt-get install python3-psycopg2 # install python3 psycopg2 module on Debian/Ubuntu - sudo yum install python-psycopg2 # install python2 psycopg2 on RedHat/Fedora/CentOS + sudo apt-get install python3-psycopg2 # install psycopg2 module on Debian/Ubuntu + sudo yum install python3-psycopg2 # install psycopg2 on RedHat/Fedora/CentOS 2. Install psycopg2 from the binary package @@ -165,10 +164,6 @@ Applications Should Not Use Superusers When connecting from an application, always use a non-superuser. Patroni requires access to the database to function properly. By using a superuser from an application, you can potentially use the entire connection pool, including the connections reserved for superusers, with the ``superuser_reserved_connections`` setting. If Patroni cannot access the Primary because the connection pool is full, behavior will be undesirable. -.. |Build Status| image:: https://travis-ci.org/zalando/patroni.svg?branch=master - :target: https://travis-ci.org/zalando/patroni -.. |Coverage Status| image:: https://coveralls.io/repos/zalando/patroni/badge.svg?branch=master - :target: https://coveralls.io/r/zalando/patroni?branch=master Testing Your HA Solution -------------------------------------- From 2be64e5131bc77d6c82b370b029d32e1ad65b5af Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 18 Aug 2023 13:36:32 +0200 Subject: [PATCH 6/6] Don't return logical slots for standby cluster (#2816) Cluster.get_replication_slots() didn't take into account that there can not be logical replication slots in a standby cluster replicas. It was only skipping logical slots for the standby_leader, but replicas were expecting that they will have to copy them over. Also on replicas in a standby cluster these logical slots were falsely added to the `_replication_slots` dict. --- patroni/dcs/__init__.py | 46 ++++++++++++++++++---------------- patroni/postgresql/__init__.py | 16 +++++++++--- patroni/postgresql/slots.py | 12 ++++++--- tests/test_slots.py | 3 +++ 4 files changed, 49 insertions(+), 28 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 279ce265..b76e1bda 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -932,8 +932,8 @@ class Cluster(NamedTuple('Cluster', """``True`` if cluster is configured to use replication slots.""" return bool(self.config and (self.config.data.get('postgresql') or {}).get('use_slots', True)) - def get_replication_slots(self, my_name: str, role: str, nofailover: bool, - major_version: int, show_error: bool = False) -> Dict[str, Dict[str, Any]]: + def get_replication_slots(self, my_name: str, role: str, nofailover: bool, major_version: int, *, + is_standby_cluster: bool = False, show_error: bool = False) -> Dict[str, Dict[str, Any]]: """Lookup configured slot names in the DCS, report issues found and merge with permanent slots. Will log an error if: @@ -945,11 +945,13 @@ class Cluster(NamedTuple('Cluster', :param role: role of this node. :param nofailover: ``True`` if this node is tagged to not be a failover candidate. :param major_version: postgresql major version. + :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from + the outside because we want to protect from the ``/config`` key removal. :param show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found. :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks. """ - slot_members: List[str] = self._get_slot_members(my_name, role) if self.use_slots else [] + slot_members: List[str] = self._get_slot_members(my_name, role) slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members} @@ -963,7 +965,7 @@ class Cluster(NamedTuple('Cluster', "; ".join(f"{', '.join(v)} map to {k}" for k, v in slot_conflicts.items() if len(v) > 1)) - permanent_slots: dict[str, Any] = self._get_permanent_slots(role, nofailover) if self.use_slots else {} + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover) disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( slots, permanent_slots, my_name, major_version) @@ -1022,33 +1024,32 @@ 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, role: str, nofailover: bool) -> Dict[str, Any]: + def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]: """Get configured permanent slot names. .. note:: - Permanent logical replication slots are only considered if ``use_slots`` configuration is enabled. Also, - only considered if *role* is ``primary`` or if it is a promotable ``replica`` -- what excludes a - ``standby_leader`` or ``replica`` with ``nofailover`` tag enabled. That combination is used for failing - over logical replication slots, and the latter nodes are not eligible for such task. + Permanent replication slots are only considered if ``use_slots`` configuration is enabled. + A node that is not supposed to become a leader (*nofailover*) will not have permanent replication slots. - Permanent physical slots are only considered if *role* is ``primary`` or ``standby_leader``, independently - if ``use_slots`` is enabled or not. That is done that way because even if Patroni itself is not using slots - to replicate among its members when ``use_slots`` is disabled, the user may still have configured Patroni to - keep permanent physical slots used out of Patroni. + In a standby cluster we only support physical replication slots. + The returned dictionary for a non-standby cluster always contains permanent logical replication slots in + order to show a warning if they are not supported by PostgreSQL before v11. + + :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from + the outside because we want to protect from the ``/config`` key removal. :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``. - or logical slots being consumed. :param nofailover: ``True`` if this node is tagged to not be a failover candidate. :returns: dictionary of permanent slot names mapped to attributes. """ - if role in ('master', 'primary', 'standby_leader'): - permanent_slots = (self.__permanent_slots - if role in ('master', 'primary') - else self.__permanent_physical_slots) - else: - permanent_slots = self.__permanent_logical_slots if not nofailover else {} - return permanent_slots + if not self.use_slots or nofailover: + return {} + + if is_standby_cluster: + return self.__permanent_physical_slots if role == 'standby_leader' else {} + + return self.__permanent_slots if role in ('master', 'primary') else self.__permanent_logical_slots def _get_slot_members(self, my_name: str, role: str) -> List[str]: """Get a list of member names that have replication slots sourcing from this node. @@ -1065,6 +1066,9 @@ class Cluster(NamedTuple('Cluster', :returns: list of member names. """ + if not self.use_slots: + return [] + if role in ('master', 'primary', 'standby_leader'): slot_members = [m.name for m in self.members if m.name != my_name diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 7f3bdccb..9b1991a0 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -429,7 +429,18 @@ class Postgresql(object): :param global_config: last known :class:`GlobalConfig` object """ self._cluster_info_state = {} - if cluster and cluster.config and cluster.config.modify_version: + + if global_config: + self._global_config = global_config + + if not self._global_config: + return + + if self._global_config.is_standby_cluster: + # 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) @@ -439,9 +450,6 @@ class Postgresql(object): self._has_permanent_logical_slots or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) - if global_config: - self._global_config = global_config - def _cluster_info_state_get(self, name: str) -> Optional[Any]: if not self._cluster_info_state: try: diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 319119ce..e4543c71 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -471,6 +471,11 @@ class SlotsHandler: elif cluster.slots and name in cluster.slots: # We want to copy only slots with feedback in a DCS create_slots.append(name) + # Slots to be copied from the primary should be removed from the *slots* structure, + # otherwise Patroni falsely assumes that they already exist. + for name in create_slots: + slots.pop(name) + error, copy_slots = self.schedule_advance_slots(advance_slots) if error: self._schedule_load_slots = True @@ -493,12 +498,13 @@ class SlotsHandler: :returns: list of logical replication slots names that should be copied from the primary. """ ret = [] - if self._postgresql.major_version >= 90400 and cluster.config: + if self._postgresql.major_version >= 90400 and self._postgresql.global_config and cluster.config: try: self.load_replication_slots() - slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role, - nofailover, self._postgresql.major_version, True) + slots = cluster.get_replication_slots( + self._postgresql.name, self._postgresql.role, nofailover, self._postgresql.major_version, + is_standby_cluster=self._postgresql.global_config.is_standby_cluster, show_error=True) self._drop_incorrect_slots(cluster, slots, paused) diff --git a/tests/test_slots.py b/tests/test_slots.py index df8c6955..add0fdf7 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -7,6 +7,7 @@ from mock import Mock, PropertyMock, patch from threading import Thread from patroni import psycopg +from patroni.config import GlobalConfig from patroni.dcs import Cluster, ClusterConfig, Member, SyncState from patroni.postgresql import Postgresql from patroni.postgresql.misc import fsync_dir @@ -28,6 +29,7 @@ class TestSlotsHandler(BaseTestPostgresql): @patch.object(Postgresql, 'is_running', Mock(return_value=True)) def setUp(self): super(TestSlotsHandler, self).setUp() + self.p._global_config = GlobalConfig({}) self.s = self.p.slots_handler self.p.start() config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1) @@ -44,6 +46,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.s.sync_replication_slots(cluster, False) self.p.set_role('standby_leader') with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \ + patch.object(GlobalConfig, 'is_standby_cluster', PropertyMock(return_value=True)), \ patch('patroni.postgresql.slots.logger.debug') as mock_debug: self.s.sync_replication_slots(cluster, False) mock_debug.assert_called_once()