From 89d794facc3e1fa728ccfbe6ac0323d668c8e021 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 24 Aug 2023 16:13:22 +0200 Subject: [PATCH 01/11] Introduce connection pool (#2829) Make it hold connection kwargs for local connections and all `NamedConnection` objects use them automatically. Also get rid of redundant `ConfigHandler.local_connect_kwargs`. On top of that we will introduce a dedicated connection for the REST API thread. --- patroni/postgresql/__init__.py | 15 ++--- patroni/postgresql/bootstrap.py | 2 +- patroni/postgresql/citus.py | 16 ++--- patroni/postgresql/config.py | 59 ++++++++++++------ patroni/postgresql/connection.py | 101 +++++++++++++++++++++++++------ patroni/postgresql/slots.py | 3 +- tests/test_bootstrap.py | 4 +- tests/test_citus.py | 2 +- tests/test_postgresql.py | 5 +- 9 files changed, 145 insertions(+), 62 deletions(-) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 9b1991a0..a37e15e1 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -18,7 +18,7 @@ from .bootstrap import Bootstrap from .callback_executor import CallbackAction, CallbackExecutor from .cancellable import CancellableSubprocess from .config import ConfigHandler, mtime -from .connection import Connection, get_connection_cursor +from .connection import ConnectionPool, get_connection_cursor from .citus import CitusHandler from .misc import parse_history, parse_lsn, postgres_major_version_to_int from .postmaster import PostmasterProcess @@ -79,7 +79,8 @@ class Postgresql(object): self.set_state('stopped') self._pending_restart = False - self._connection = Connection() + self.connection_pool = ConnectionPool() + self._connection = self.connection_pool.get('heartbeat') self.citus_handler = CitusHandler(self, config.get('citus')) self.config = ConfigHandler(self, config) self.config.check_directories() @@ -277,7 +278,7 @@ class Postgresql(object): :returns: 'ok' if PostgreSQL is up, 'reject' if starting up, 'no_resopnse' if not up.""" - r = self.config.local_connect_kwargs + r = self.connection_pool.conn_kwargs cmd = [self.pgcommand('pg_isready'), '-p', r['port'], '-d', self._database] # Host is not set if we are connecting via default unix socket @@ -328,10 +329,6 @@ class Postgresql(object): def connection(self) -> Union['connection3', 'Connection3[Any]']: return self._connection.get() - def set_connection_kwargs(self, kwargs: Dict[str, Any]) -> None: - self._connection.set_conn_kwargs(kwargs.copy()) - self.citus_handler.set_conn_kwargs(kwargs.copy()) - def _query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: """Execute *sql* query with *params* and optionally return results. @@ -694,7 +691,7 @@ class Postgresql(object): # the former node, otherwise, we might get a stalled one # after kill -9, which would report incorrect data to # patroni. - self._connection.close() + self.connection_pool.close() if self.is_running(): logger.error('Cannot start PostgreSQL because one is already running.') @@ -765,7 +762,7 @@ class Postgresql(object): def checkpoint(self, connect_kwargs: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> Optional[str]: check_not_is_in_recovery = connect_kwargs is not None - connect_kwargs = connect_kwargs or self.config.local_connect_kwargs + connect_kwargs = connect_kwargs or self.connection_pool.conn_kwargs for p in ['connect_timeout', 'options']: connect_kwargs.pop(p, None) if timeout: diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py index 6e25012b..26025e43 100644 --- a/patroni/postgresql/bootstrap.py +++ b/patroni/postgresql/bootstrap.py @@ -176,7 +176,7 @@ class Bootstrap(object): """ cmd = config.get('post_bootstrap') or config.get('post_init') if cmd: - r = self._postgresql.config.local_connect_kwargs + r = self._postgresql.connection_pool.conn_kwargs connstring = self._postgresql.config.format_dsn(r, True) if 'host' not in r: # https://www.postgresql.org/docs/current/static/libpq-pgpass.html diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index f659e325..09f77f2b 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -6,7 +6,6 @@ from threading import Condition, Event, Thread from urllib.parse import urlparse from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING -from .connection import Connection from ..dcs import CITUS_COORDINATOR_GROUP_ID, Cluster from ..psycopg import connect, quote_ident @@ -71,7 +70,10 @@ class CitusHandler(Thread): self.daemon = True self._postgresql = postgresql self._config = config - self._connection = Connection() + if config: + self._connection = postgresql.connection_pool.get( + 'citus', {'dbname': config['database'], + 'options': '-c statement_timeout=0 -c idle_in_transaction_session_timeout=0'}) self._pg_dist_node: Dict[int, PgDistNode] = {} # Cache of pg_dist_node: {groupid: PgDistNode()} self._tasks: List[PgDistNode] = [] # Requests to change pg_dist_node, every task is a `PgDistNode` self._in_flight: Optional[PgDistNode] = None # Reference to the `PgDistNode` being changed in a transaction @@ -91,12 +93,6 @@ class CitusHandler(Thread): def is_worker(self) -> bool: return self.is_enabled() and not self.is_coordinator() - def set_conn_kwargs(self, kwargs: Dict[str, Any]) -> None: - if isinstance(self._config, dict): # self.is_enabled(): - kwargs.update({'dbname': self._config['database'], - 'options': '-c statement_timeout=0 -c idle_in_transaction_session_timeout=0'}) - self._connection.set_conn_kwargs(kwargs) - def schedule_cache_rebuild(self) -> None: with self._condition: self._schedule_load_pg_dist_node = True @@ -359,8 +355,8 @@ class CitusHandler(Thread): if not isinstance(self._config, dict): # self.is_enabled() return - conn_kwargs = self._postgresql.config.local_connect_kwargs - conn_kwargs['options'] = '-c synchronous_commit=local -c statement_timeout=0' + conn_kwargs = {**self._postgresql.connection_pool.conn_kwargs, + 'options': '-c synchronous_commit=local -c statement_timeout=0'} if self._config['database'] != self._postgresql.database: conn = connect(**conn_kwargs) try: diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 51bbf9af..79141256 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -942,24 +942,32 @@ class ConfigHandler(object): return 'localhost' # connection via localhost is preferred return listen_addresses[0].strip() # can't use localhost, take first address from listen_addresses - @property - def local_connect_kwargs(self) -> Dict[str, Any]: - ret = self._local_address.copy() - # add all of the other connection settings that are available - ret.update(self._superuser) - # if the "username" parameter is present, it actually needs to be "user" - # for connecting to PostgreSQL - if 'username' in self._superuser: - ret['user'] = self._superuser['username'] - del ret['username'] - # ensure certain Patroni configurations are available - ret.update({'dbname': self._postgresql.database, - 'fallback_application_name': 'Patroni', - 'connect_timeout': 3, - 'options': '-c statement_timeout=2000'}) - return ret - def resolve_connection_addresses(self) -> None: + """Calculates and sets local and remote connection urls and options. + + This method sets: + * :attr:`Postgresql.connection_string ` attribute, which + is later written to the member key in DCS as ``conn_url``. + * :attr:`ConfigHandler.local_replication_address` attribute, which is used for replication connections to + local postgres. + * :attr:`ConnectionPool.conn_kwargs ` attribute, + which is used for superuser connections to local postgres. + + .. note:: + If there is a valid directory in ``postgresql.parameters.unix_socket_directories`` in the Patroni + configuration and ``postgresql.use_unix_socket`` and/or ``postgresql.use_unix_socket_repl`` + are set to ``True``, we respectively use unix sockets for superuser and replication connections + to local postgres. + + If there is a requirement to use unix sockets, but nothing is set in the + ``postgresql.parameters.unix_socket_directories``, we omit a ``host`` in connection parameters relying + on the ability of ``libpq`` to connect via some default unix socket directory. + + If unix sockets are not requested we "switch" to TCP, prefering to use ``localhost`` if it is possible + to deduce that Postgres is listening on a local interface address. + + Otherwise we just used the first address specified in the ``listen_addresses`` GUC. + """ port = self._server_parameters['port'] tcp_local_address = self._get_tcp_local_address() netloc = self._config.get('connect_address') or tcp_local_address + ':' + port @@ -972,12 +980,25 @@ class ConfigHandler(object): tcp_local_address = {'host': tcp_local_address, 'port': port} - self._local_address = unix_local_address if self._config.get('use_unix_socket') else tcp_local_address self.local_replication_address = unix_local_address\ if self._config.get('use_unix_socket_repl') else tcp_local_address self._postgresql.connection_string = uri('postgres', netloc, self._postgresql.database) - self._postgresql.set_connection_kwargs(self.local_connect_kwargs) + + local_address = unix_local_address if self._config.get('use_unix_socket') else tcp_local_address + local_conn_kwargs = { + **local_address, + **self._superuser, + 'dbname': self._postgresql.database, + 'fallback_application_name': 'Patroni', + 'connect_timeout': 3, + 'options': '-c statement_timeout=2000' + } + # if the "username" parameter is present, it actually needs to be "user" for connecting to PostgreSQL + if 'username' in local_conn_kwargs: + local_conn_kwargs['user'] = local_conn_kwargs.pop('username') + # "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] diff --git a/patroni/postgresql/connection.py b/patroni/postgresql/connection.py index 5bb97a1f..2a50dbb5 100644 --- a/patroni/postgresql/connection.py +++ b/patroni/postgresql/connection.py @@ -2,9 +2,9 @@ import logging from contextlib import contextmanager from threading import Lock -from typing import Any, Dict, Iterator, List, Union, Tuple, TYPE_CHECKING +from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING if TYPE_CHECKING: # pragma: no cover - from psycopg import Connection as Connection3, Cursor + from psycopg import Connection, Cursor from psycopg2 import connection, cursor from .. import psycopg @@ -13,27 +13,34 @@ from ..exceptions import PostgresConnectionException logger = logging.getLogger(__name__) -class Connection: - """Helper class to manage connections from Patroni to PostgreSQL. +class NamedConnection: + """Helper class to manage ``psycopg`` 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: - """Create an instance of :class:`Connection` class.""" + def __init__(self, pool: 'ConnectionPool', name: str, kwargs_override: Optional[Dict[str, Any]]) -> None: + """Create an instance of :class:`NamedConnection` class. + + :param pool: reference to a :class:`ConnectionPool` object. + :param name: name of the connection. + :param kwargs_override: :class:`dict` object with connection parameters that should be + different from default values provided by connection *pool*. + """ + self._pool = pool + self._name = name + self._kwargs_override = kwargs_override or {} self._lock = Lock() # used to make sure that only one connection to postgres is established self._connection = None - def set_conn_kwargs(self, conn_kwargs: Dict[str, Any]) -> None: - """Set connection parameters, like user, password, host, port and so on. + @property + def _conn_kwargs(self) -> Dict[str, Any]: + """Connection parameters for this :class:`NamedConnection`.""" + return {**self._pool.conn_kwargs, **self._kwargs_override, 'application_name': f'Patroni {self._name}'} - :param conn_kwargs: connection parameters as a dictionary. - """ - self._conn_kwargs = conn_kwargs - - def get(self) -> Union['connection', 'Connection3[Any]']: + def get(self) -> Union['connection', 'Connection[Any]']: """Get ``psycopg``/``psycopg2`` connection object. .. note:: @@ -43,7 +50,7 @@ class Connection: """ with self._lock: if not self._connection or self._connection.closed != 0: - logger.info("establishing a new patroni connection to postgres") + logger.info("establishing a new patroni %s connection to postgres", self._name) self._connection = psycopg.connect(**self._conn_kwargs) self.server_version = getattr(self._connection, 'server_version', 0) return self._connection @@ -76,12 +83,72 @@ class Connection: raise exc raise PostgresConnectionException('connection problems') from exc - def close(self) -> None: - """Close the psycopg connection to postgres.""" + def close(self, silent: bool = False) -> bool: + """Close the psycopg connection to postgres. + + :param silent: whether the method should not write logs. + + :returns: ``True`` if ``psycopg`` connection was closed, ``False`` otherwise.`` + """ + ret = False if self._connection and self._connection.closed == 0: self._connection.close() - logger.info("closed patroni connection to postgres") + if not silent: + logger.info("closed patroni %s connection to postgres", self._name) + ret = True self._connection = None + return ret + + +class ConnectionPool: + """Helper class to manage named connections from Patroni to PostgreSQL. + + The instance keeps named :class:`NamedConnection` objects and parameters that must be used for new connections. + """ + + def __init__(self) -> None: + """Create an instance of :class:`ConnectionPool` class.""" + self._lock = Lock() + self._connections: Dict[str, NamedConnection] = {} + self._conn_kwargs: Dict[str, Any] = {} + + @property + def conn_kwargs(self) -> Dict[str, Any]: + """Connection parameters that must be used for new ``psycopg`` connections.""" + with self._lock: + return self._conn_kwargs.copy() + + @conn_kwargs.setter + def conn_kwargs(self, value: Dict[str, Any]) -> None: + """Set new connection parameters. + + :param value: :class:`dict` object with connection parameters. + """ + with self._lock: + self._conn_kwargs = value + + def get(self, name: str, kwargs_override: Optional[Dict[str, Any]] = None) -> NamedConnection: + """Get a new named :class:`NamedConnection` object from the pool. + + .. note:: + Creates a new :class:`NamedConnection` object if it doesn't yet exist in the pool. + + :param name: name of the connection. + :param kwargs_override: :class:`dict` object with connection parameters that should be + different from default values provided by :attr:`conn_kwargs`. + + :returns: :class:`NamedConnection` object. + """ + with self._lock: + if name not in self._connections: + self._connections[name] = NamedConnection(self, name, kwargs_override) + return self._connections[name] + + def close(self) -> None: + """Close all named connections from Patroni to PostgreSQL registered in the pool.""" + with self._lock: + if any(conn.close(True) for conn in self._connections.values()): + logger.info("closed patroni connections to postgres") @contextmanager diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index e4543c71..51a8fc5a 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -384,8 +384,7 @@ class SlotsHandler: :yields: connection cursor object, note implementation varies depending on version of :mod:`psycopg`. """ - conn_kwargs = self._postgresql.config.local_connect_kwargs - conn_kwargs.update(kwargs) + conn_kwargs = {**self._postgresql.connection_pool.conn_kwargs, **kwargs} with get_connection_cursor(**conn_kwargs) as cur: yield cur diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index 9f98fecb..c922fcae 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -250,7 +250,7 @@ class TestBootstrap(BaseTestPostgresql): self.assertFalse(self.b.call_post_bootstrap({'post_init': '/bin/false'})) mock_cancellable_subprocess_call.return_value = 0 - self.p.config.superuser.pop('username') + self.p.connection_pool._conn_kwargs.pop('user') self.assertTrue(self.b.call_post_bootstrap({'post_init': '/bin/false'})) mock_cancellable_subprocess_call.assert_called() args, kwargs = mock_cancellable_subprocess_call.call_args @@ -258,7 +258,7 @@ class TestBootstrap(BaseTestPostgresql): self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=127.0.0.2 port=5432']) mock_cancellable_subprocess_call.reset_mock() - self.p.config._local_address.pop('host') + self.p.connection_pool._conn_kwargs.pop('host') self.assertTrue(self.b.call_post_bootstrap({'post_init': '/bin/false'})) mock_cancellable_subprocess_call.assert_called() self.assertEqual(mock_cancellable_subprocess_call.call_args[0][0], ['/bin/false', 'dbname=postgres port=5432']) diff --git a/tests/test_citus.py b/tests/test_citus.py index 7c2d63bb..9849a069 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -13,7 +13,7 @@ class TestCitus(BaseTestPostgresql): def setUp(self): super(TestCitus, self).setUp() self.c = self.p.citus_handler - self.c.set_conn_kwargs({'host': 'localhost', 'dbname': 'postgres'}) + self.p.connection_pool.conn_kwargs = {'host': 'localhost', 'dbname': 'postgres'} self.cluster = get_cluster_initialized_with_leader() self.cluster.workers[1] = self.cluster diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 510c2a90..29b25ee4 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -577,7 +577,10 @@ class TestPostgresql(BaseTestPostgresql): self.assertEqual(self.p.config.local_replication_address, {'host': '/tmp', 'port': '5432'}) self.p.config._server_parameters.pop('unix_socket_directories') self.p.config.resolve_connection_addresses() - self.assertEqual(self.p.config._local_address, {'port': '5432'}) + self.assertEqual(self.p.connection_pool.conn_kwargs, {'connect_timeout': 3, 'dbname': 'postgres', + 'fallback_application_name': 'Patroni', + 'options': '-c statement_timeout=2000', + 'password': 'test', 'port': '5432', 'user': 'foo'}) @patch.object(Postgresql, '_version_file_exists', Mock(return_value=True)) def test_get_major_version(self): From 77dba39585cf1baa25852bc002bb4522b16bc516 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 4 Sep 2023 09:00:24 +0200 Subject: [PATCH 02/11] Pin version of sphinx-github-style (#2847) 1.0.3 removed support of `top_level` configuration parameter and builds now are failing. Besides that remove redundant pyyaml from requirements.docs.txt --- requirements.docs.txt | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/requirements.docs.txt b/requirements.docs.txt index 1a71f097..afb47378 100644 --- a/requirements.docs.txt +++ b/requirements.docs.txt @@ -1,5 +1,4 @@ sphinx>=4 sphinx_rtd_theme>1 sphinxcontrib-apidoc -sphinx-github-style -pyyaml +sphinx-github-style<1.0.3 From 03107e6d8b7660ae1a05bf3c91374681bfdb096c Mon Sep 17 00:00:00 2001 From: Israel Date: Mon, 4 Sep 2023 04:27:46 -0300 Subject: [PATCH 03/11] `patronictl --help` was showing `ctl` function's docstring (#2845) `patronictl` is implemented using `click` module, and that module uses the functions' docstrings for creating a helper text. As a consequence the docstring for `ctl` function was being shown to the user, which doesn't make sense. This PR fixes that issue by adding a user-friendly description to be shown on `patronictl --help`. We use a `\f` to tell `click` when to stop capturing text to show in the helper. Note that `patronictl` commands are implemented using `@ctl.command` decorator, and we always provide them with `help` argument. That said, none of the subcommands are affected by the aforementioned issue, only the entry point of the CLI. References: PAT-201. --- patroni/ctl.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index b1d63d2b..c8a396c9 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -264,7 +264,9 @@ role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 's @click.option('-k', '--insecure', is_flag=True, help='Allow connections to SSL sites without certs') @click.pass_context def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: bool) -> None: - """Entry point of ``patronictl`` utility. + """Command-line interface for interacting with Patroni. + \f + Entry point of ``patronictl`` utility. Load the configuration file. From 6b7f914da7fd4974ca091c8f55c769b2ad0f58a3 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 4 Sep 2023 10:03:37 +0200 Subject: [PATCH 04/11] Fix bug with kubernetes.standby_leader_label_value (#2832) When running with the leader lock Patroni was just setting the `role` label to `master` and effectively `kubernetes.standby_leader_label_value` feature never worked. Now it is fixed, but in order to not introduce breaking changes we just update default value of the `standby_leader_label_value` to the `master`. --- docs/ENVIRONMENT.rst | 2 +- docs/yaml_configuration.rst | 2 +- patroni/dcs/kubernetes.py | 9 +++------ tests/test_kubernetes.py | 12 +++++++----- 4 files changed, 12 insertions(+), 13 deletions(-) diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index 00595cf7..e3762aca 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -115,7 +115,7 @@ Kubernetes - **PATRONI\_KUBERNETES\_ROLE\_LABEL**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``. - **PATRONI\_KUBERNETES\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `master`. Default value is `master`. - **PATRONI\_KUBERNETES\_FOLLOWER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `replica`. Default value is `replica`. -- **PATRONI\_KUBERNETES\_STANDBY\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``. +- **PATRONI\_KUBERNETES\_STANDBY\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is ``standby_leader``. Default value is ``master``. - **PATRONI\_KUBERNETES\_TMP\_ROLE\_LABEL**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary. - **PATRONI\_KUBERNETES\_USE\_ENDPOINTS**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state. - **PATRONI\_KUBERNETES\_POD\_IP**: (optional) IP address of the pod Patroni is running in. This value is required when `PATRONI_KUBERNETES_USE_ENDPOINTS` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted. diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst index f0cb8ef4..d9283192 100644 --- a/docs/yaml_configuration.rst +++ b/docs/yaml_configuration.rst @@ -171,7 +171,7 @@ Kubernetes - **role\_label**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``. - **leader\_label\_value**: (optional) value of the pod label when Postgres role is ``master``. Default value is ``master``. - **follower\_label\_value**: (optional) value of the pod label when Postgres role is ``replica``. Default value is ``replica``. -- **standby\_leader\_label\_value**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``. +- **standby\_leader\_label\_value**: (optional) value of the pod label when Postgres role is ``standby_leader``. Default value is ``master``. - **tmp_\role\_label**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary. - **use\_endpoints**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state. - **pod\_ip**: (optional) IP address of the pod Patroni is running in. This value is required when `use_endpoints` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted. diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index fec544b0..47ba4e7a 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -756,7 +756,7 @@ class Kubernetes(AbstractDCS): self._role_label = config.get('role_label', 'role') self._leader_label_value = config.get('leader_label_value', 'master') self._follower_label_value = config.get('follower_label_value', 'replica') - self._standby_leader_label_value = config.get('standby_leader_label_value', 'standby-leader') + self._standby_leader_label_value = config.get('standby_leader_label_value', 'master') self._tmp_role_label = config.get('tmp_role_label') self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME super(Kubernetes, self).__init__({**config, 'namespace': ''}) @@ -1269,13 +1269,10 @@ class Kubernetes(AbstractDCS): def touch_member(self, data: Dict[str, Any]) -> bool: cluster = self.cluster if cluster and cluster.leader and cluster.leader.name == self._name: - role = self._leader_label_value + role = self._standby_leader_label_value if data['role'] == 'standby_leader' else self._leader_label_value tmp_role = 'master' elif data['state'] == 'running' and data['role'] not in ('master', 'primary'): - role = { - 'replica': self._follower_label_value, - 'standby-leader': self._standby_leader_label_value, - }.get(data['role'], data['role']) + role = {'replica': self._follower_label_value}.get(data['role'], data['role']) tmp_role = data['role'] else: role = None diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 694cd505..c33bbecc 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -308,13 +308,15 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): mock_patch_namespaced_pod.assert_called() self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'replica') - - self.k.touch_member({'state': 'running', 'role': 'standby-leader'}) - mock_patch_namespaced_pod.assert_called() - self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') - self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'standby-leader') + mock_patch_namespaced_pod.rest_mock() self.k._name = 'p-0' + self.k.touch_member({'role': 'standby_leader'}) + mock_patch_namespaced_pod.assert_called() + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'master') + mock_patch_namespaced_pod.rest_mock() + self.k.touch_member({'role': 'primary'}) mock_patch_namespaced_pod.assert_called() self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'true') From d2603402ea0a02aefd00ee709c5343b4ab0a4beb Mon Sep 17 00:00:00 2001 From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com> Date: Mon, 4 Sep 2023 20:24:26 +0100 Subject: [PATCH 05/11] Debian docker image pip error (#2849) * Use virtualenv to install tox in behave Dockerfile Upstream change in postgres docker image uses debian restriction on installing system-wide non-debian python packages. Debian doesn't provide a tox>=4, so we need to install with pip. * Exclude all output directories generated using `tox-wrapper.sh` The `tox-wrapper.sh` script created by `features/Dockerfile` creates directories like features/output-tox-pg14-docker-behave-etcd-lin-973719674/ * Reduce footprint of tox behave docker image --- .gitignore | 2 +- features/Dockerfile | 7 +++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/.gitignore b/.gitignore index 23cbf17a..c902c6eb 100644 --- a/.gitignore +++ b/.gitignore @@ -33,7 +33,7 @@ nosetests.xml coverage.xml htmlcov junit.xml -features/output +features/output* dummy # Translations diff --git a/features/Dockerfile b/features/Dockerfile index 86c6041d..7f52c2af 100644 --- a/features/Dockerfile +++ b/features/Dockerfile @@ -27,8 +27,8 @@ RUN set -ex \ && apt-get update \ && apt-get reinstall init-system-helpers \ && apt-get install -y \ - python3-pip \ python3-dev \ + python3-venv \ rsync \ curl \ gcc \ @@ -40,7 +40,9 @@ RUN set -ex \ net-tools \ iputils-ping \ && rm -rf /var/cache/apt \ - && python3 -m pip install --no-cache-dir tox \ + \ + && python3 -m venv /tox \ + && /tox/bin/pip install --no-cache-dir tox>=4 \ \ && mkdir -p "$PGHOME" \ && sed -i "s|/var/lib/postgresql.*|$PGHOME:/bin/bash|" /etc/passwd \ @@ -50,6 +52,7 @@ RUN set -ex \ && curl -sL "$ETCDURL/etcd-v$ETCDVERSION-linux-$(dpkg --print-architecture).tar.gz" \ | tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl +ENV PATH="/tox/bin:$PATH" # This Dockerfile syntax only works with docker buildx and the syntax # line at the top of this file. From 80a03a4892bfa12c8a87862d0c975390da1365cd Mon Sep 17 00:00:00 2001 From: SK <78915702+sskserk@users.noreply.github.com> Date: Tue, 5 Sep 2023 07:24:17 +0200 Subject: [PATCH 06/11] Enreach some endpoints with the scope and name (#2846) - monitoring endpoints - added `name` to the `patroni`, next to the `scope` and `version` - metrics endpoint - added name to labels --- docs/rest_api.rst | 60 ++++++++++++++++++++++++++--------------------- patroni/api.py | 57 ++++++++++++++++++++++++-------------------- 2 files changed, 65 insertions(+), 52 deletions(-) diff --git a/docs/rest_api.rst b/docs/rest_api.rst index 6e97ade0..e49f6ee0 100644 --- a/docs/rest_api.rst +++ b/docs/rest_api.rst @@ -131,7 +131,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -178,7 +179,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -223,7 +225,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -267,7 +270,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -279,70 +283,70 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e # HELP patroni_version Patroni semver without periods. \ # TYPE patroni_version gauge - patroni_version{scope="batman"} 020103 + patroni_version{scope="batman",name="patroni1"} 020103 # HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise. # TYPE patroni_postgres_running gauge - patroni_postgres_running{scope="batman"} 1 + patroni_postgres_running{scope="batman",name="patroni1"} 1 # HELP patroni_postmaster_start_time Epoch seconds since Postgres started. # TYPE patroni_postmaster_start_time gauge - patroni_postmaster_start_time{scope="batman"} 1657656955.179243 + patroni_postmaster_start_time{scope="batman",name="patroni1"} 1657656955.179243 # HELP patroni_master Value is 1 if this node is the leader, 0 otherwise. # TYPE patroni_master gauge - patroni_master{scope="batman"} 1 + patroni_master{scope="batman",name="patroni1"} 1 # HELP patroni_primary Value is 1 if this node is the leader, 0 otherwise. # TYPE patroni_primary gauge - patroni_primary{scope="batman"} 1 + patroni_primary{scope="batman",name="patroni1"} 1 # HELP patroni_xlog_location Current location of the Postgres transaction log, 0 if this node is not the leader. # TYPE patroni_xlog_location counter - patroni_xlog_location{scope="batman"} 22320573386952 + patroni_xlog_location{scope="batman",name="patroni1"} 22320573386952 # HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise. # TYPE patroni_standby_leader gauge - patroni_standby_leader{scope="batman"} 0 + patroni_standby_leader{scope="batman",name="patroni1"} 0 # HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise. # TYPE patroni_replica gauge - patroni_replica{scope="batman"} 0 + patroni_replica{scope="batman",name="patroni1"} 0 # HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise. # TYPE patroni_sync_standby gauge - patroni_sync_standby{scope="batman"} 0 + patroni_sync_standby{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_received_location Current location of the received Postgres transaction log, 0 if this node is not a replica. # TYPE patroni_xlog_received_location counter - patroni_xlog_received_location{scope="batman"} 0 + patroni_xlog_received_location{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_replayed_location Current location of the replayed Postgres transaction log, 0 if this node is not a replica. # TYPE patroni_xlog_replayed_location counter - patroni_xlog_replayed_location{scope="batman"} 0 + patroni_xlog_replayed_location{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed Postgres transaction log, 0 if null. # TYPE patroni_xlog_replayed_timestamp gauge - patroni_xlog_replayed_timestamp{scope="batman"} 0 + patroni_xlog_replayed_timestamp{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise. # TYPE patroni_xlog_paused gauge - patroni_xlog_paused{scope="batman"} 0 + patroni_xlog_paused{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise. # TYPE patroni_postgres_streaming gauge - patroni_postgres_streaming{scope="batman"} 1 + patroni_postgres_streaming{scope="batman",name="patroni1"} 1 # HELP patroni_postgres_in_archive_recovery Value is 1 if Postgres is replicating from archive, 0 otherwise. # TYPE patroni_postgres_in_archive_recovery gauge - patroni_postgres_in_archive_recovery{scope="batman"} 0 + patroni_postgres_in_archive_recovery{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise. # TYPE patroni_postgres_server_version gauge - patroni_postgres_server_version {scope="batman"} 140004 + patroni_postgres_server_version{scope="batman",name="patroni1"} 140004 # HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked. # TYPE patroni_cluster_unlocked gauge - patroni_cluster_unlocked{scope="batman"} 0 + patroni_cluster_unlocked{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise. # TYPE patroni_postgres_timeline counter - patroni_failsafe_mode_is_active{scope="batman"} 0 + patroni_failsafe_mode_is_active{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise. # TYPE patroni_postgres_timeline counter - patroni_postgres_timeline{scope="batman"} 24 + patroni_postgres_timeline{scope="batman",name="patroni1"} 24 # HELP patroni_dcs_last_seen Epoch timestamp when DCS was last contacted successfully by Patroni. # TYPE patroni_dcs_last_seen gauge - patroni_dcs_last_seen{scope="batman"} 1677658321 + patroni_dcs_last_seen{scope="batman",name="patroni1"} 1677658321 # HELP patroni_pending_restart Value is 1 if the node needs a restart, 0 otherwise. # TYPE patroni_pending_restart gauge - patroni_pending_restart{scope="batman"} 1 + patroni_pending_restart{scope="batman",name="patroni1"} 1 # HELP patroni_is_paused Value is 1 if auto failover is disabled, 0 otherwise. # TYPE patroni_is_paused gauge - patroni_is_paused{scope="batman"} 1 + patroni_is_paused{scope="batman",name="patroni1"} 1 Cluster status endpoints @@ -381,6 +385,7 @@ Cluster status endpoints "lag": 0 } ], + "scope": "demo", "scheduled_switchover": { "at": "2023-09-24T10:36:00+02:00", "from": "patroni1", @@ -489,8 +494,9 @@ Let's check that the node processed this configuration. First of all it should s "location": 2197818976 }, "patroni": { + "version": "1.0", "scope": "batman", - "version": "1.0" + "name": "patroni1" }, "state": "running", "role": "master", diff --git a/patroni/api.py b/patroni/api.py index b01c24d8..a7a754a3 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -198,7 +198,11 @@ class RestApiHandler(BaseHTTPRequestHandler): response['database_system_identifier'] = patroni.postgresql.sysid if patroni.postgresql.pending_restart: response['pending_restart'] = True - response['patroni'] = {'version': patroni.version, 'scope': patroni.postgresql.scope} + response['patroni'] = { + 'version': patroni.version, + 'scope': patroni.postgresql.scope, + 'name': patroni.postgresql.name + } if patroni.scheduled_restart: response['scheduled_restart'] = patroni.scheduled_restart.copy() del response['scheduled_restart']['postmaster_start_time'] @@ -449,7 +453,10 @@ class RestApiHandler(BaseHTTPRequestHandler): """ cluster = self.server.patroni.dcs.get_cluster(True) global_config = self.server.patroni.config.get_global_config(cluster) - self._write_json_response(200, cluster_as_json(cluster, global_config)) + + response = cluster_as_json(cluster, global_config) + response['scope'] = self.server.patroni.postgresql.scope + self._write_json_response(200, response) def do_GET_history(self) -> None: """Handle a ``GET`` request to ``/history`` path. @@ -526,113 +533,113 @@ class RestApiHandler(BaseHTTPRequestHandler): metrics: List[str] = [] - scope_label = '{{scope="{0}"}}'.format(patroni.postgresql.scope) + labels = f'{{scope="{patroni.postgresql.scope}",name="{patroni.postgresql.name}"}}' metrics.append("# HELP patroni_version Patroni semver without periods.") metrics.append("# TYPE patroni_version gauge") padded_semver = ''.join([x.zfill(2) for x in patroni.version.split('.')]) # 2.0.2 => 020002 - metrics.append("patroni_version{0} {1}".format(scope_label, padded_semver)) + metrics.append("patroni_version{0} {1}".format(labels, padded_semver)) metrics.append("# HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise.") metrics.append("# TYPE patroni_postgres_running gauge") - metrics.append("patroni_postgres_running{0} {1}".format(scope_label, int(postgres['state'] == 'running'))) + metrics.append("patroni_postgres_running{0} {1}".format(labels, int(postgres['state'] == 'running'))) metrics.append("# HELP patroni_postmaster_start_time Epoch seconds since Postgres started.") metrics.append("# TYPE patroni_postmaster_start_time gauge") postmaster_start_time = postgres.get('postmaster_start_time') postmaster_start_time = (postmaster_start_time - epoch).total_seconds() if postmaster_start_time else 0 - metrics.append("patroni_postmaster_start_time{0} {1}".format(scope_label, postmaster_start_time)) + metrics.append("patroni_postmaster_start_time{0} {1}".format(labels, postmaster_start_time)) metrics.append("# HELP patroni_master Value is 1 if this node is the leader, 0 otherwise.") metrics.append("# TYPE patroni_master gauge") - metrics.append("patroni_master{0} {1}".format(scope_label, int(postgres['role'] in ('master', 'primary')))) + metrics.append("patroni_master{0} {1}".format(labels, int(postgres['role'] in ('master', 'primary')))) metrics.append("# HELP patroni_primary Value is 1 if this node is the leader, 0 otherwise.") metrics.append("# TYPE patroni_primary gauge") - metrics.append("patroni_primary{0} {1}".format(scope_label, int(postgres['role'] in ('master', 'primary')))) + metrics.append("patroni_primary{0} {1}".format(labels, int(postgres['role'] in ('master', 'primary')))) metrics.append("# HELP patroni_xlog_location Current location of the Postgres" " transaction log, 0 if this node is not the leader.") metrics.append("# TYPE patroni_xlog_location counter") - metrics.append("patroni_xlog_location{0} {1}".format(scope_label, postgres.get('xlog', {}).get('location', 0))) + metrics.append("patroni_xlog_location{0} {1}".format(labels, postgres.get('xlog', {}).get('location', 0))) metrics.append("# HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise.") metrics.append("# TYPE patroni_standby_leader gauge") - metrics.append("patroni_standby_leader{0} {1}".format(scope_label, int(postgres['role'] == 'standby_leader'))) + metrics.append("patroni_standby_leader{0} {1}".format(labels, int(postgres['role'] == 'standby_leader'))) metrics.append("# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.") metrics.append("# TYPE patroni_replica gauge") - metrics.append("patroni_replica{0} {1}".format(scope_label, int(postgres['role'] == 'replica'))) + metrics.append("patroni_replica{0} {1}".format(labels, int(postgres['role'] == 'replica'))) metrics.append("# HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise.") metrics.append("# TYPE patroni_sync_standby gauge") - metrics.append("patroni_sync_standby{0} {1}".format(scope_label, int(postgres.get('sync_standby', False)))) + metrics.append("patroni_sync_standby{0} {1}".format(labels, int(postgres.get('sync_standby', False)))) metrics.append("# HELP patroni_xlog_received_location Current location of the received" " Postgres transaction log, 0 if this node is not a replica.") metrics.append("# TYPE patroni_xlog_received_location counter") metrics.append("patroni_xlog_received_location{0} {1}" - .format(scope_label, postgres.get('xlog', {}).get('received_location', 0))) + .format(labels, postgres.get('xlog', {}).get('received_location', 0))) metrics.append("# HELP patroni_xlog_replayed_location Current location of the replayed" " Postgres transaction log, 0 if this node is not a replica.") metrics.append("# TYPE patroni_xlog_replayed_location counter") metrics.append("patroni_xlog_replayed_location{0} {1}" - .format(scope_label, postgres.get('xlog', {}).get('replayed_location', 0))) + .format(labels, postgres.get('xlog', {}).get('replayed_location', 0))) metrics.append("# HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed" " Postgres transaction log, 0 if null.") metrics.append("# TYPE patroni_xlog_replayed_timestamp gauge") replayed_timestamp = postgres.get('xlog', {}).get('replayed_timestamp') replayed_timestamp = (replayed_timestamp - epoch).total_seconds() if replayed_timestamp else 0 - metrics.append("patroni_xlog_replayed_timestamp{0} {1}".format(scope_label, replayed_timestamp)) + metrics.append("patroni_xlog_replayed_timestamp{0} {1}".format(labels, replayed_timestamp)) metrics.append("# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.") metrics.append("# TYPE patroni_xlog_paused gauge") metrics.append("patroni_xlog_paused{0} {1}" - .format(scope_label, int(postgres.get('xlog', {}).get('paused', False) is True))) + .format(labels, int(postgres.get('xlog', {}).get('paused', False) is True))) if postgres.get('server_version', 0) >= 90600: metrics.append("# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.") metrics.append("# TYPE patroni_postgres_streaming gauge") metrics.append("patroni_postgres_streaming{0} {1}" - .format(scope_label, int(postgres.get('replication_state') == 'streaming'))) + .format(labels, int(postgres.get('replication_state') == 'streaming'))) metrics.append("# HELP patroni_postgres_in_archive_recovery Value is 1" " if Postgres is replicating from archive, 0 otherwise.") metrics.append("# TYPE patroni_postgres_in_archive_recovery gauge") metrics.append("patroni_postgres_in_archive_recovery{0} {1}" - .format(scope_label, int(postgres.get('replication_state') == 'in archive recovery'))) + .format(labels, int(postgres.get('replication_state') == 'in archive recovery'))) metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.") metrics.append("# TYPE patroni_postgres_server_version gauge") - metrics.append("patroni_postgres_server_version {0} {1}".format(scope_label, postgres.get('server_version', 0))) + metrics.append("patroni_postgres_server_version {0} {1}".format(labels, postgres.get('server_version', 0))) metrics.append("# HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked.") metrics.append("# TYPE patroni_cluster_unlocked gauge") - metrics.append("patroni_cluster_unlocked{0} {1}".format(scope_label, int(postgres.get('cluster_unlocked', 0)))) + metrics.append("patroni_cluster_unlocked{0} {1}".format(labels, int(postgres.get('cluster_unlocked', 0)))) metrics.append("# HELP patroni_failsafe_mode_is_active Value is 1 if failsafe mode is active, 0 if inactive.") metrics.append("# TYPE patroni_failsafe_mode_is_active gauge") metrics.append("patroni_failsafe_mode_is_active{0} {1}" - .format(scope_label, int(postgres.get('failsafe_mode_is_active', 0)))) + .format(labels, int(postgres.get('failsafe_mode_is_active', 0)))) metrics.append("# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.") metrics.append("# TYPE patroni_postgres_timeline counter") - metrics.append("patroni_postgres_timeline{0} {1}".format(scope_label, postgres.get('timeline', 0))) + metrics.append("patroni_postgres_timeline{0} {1}".format(labels, postgres.get('timeline', 0))) metrics.append("# HELP patroni_dcs_last_seen Epoch timestamp when DCS was last contacted successfully" " by Patroni.") metrics.append("# TYPE patroni_dcs_last_seen gauge") - metrics.append("patroni_dcs_last_seen{0} {1}".format(scope_label, postgres.get('dcs_last_seen', 0))) + metrics.append("patroni_dcs_last_seen{0} {1}".format(labels, postgres.get('dcs_last_seen', 0))) metrics.append("# HELP patroni_pending_restart Value is 1 if the node needs a restart, 0 otherwise.") metrics.append("# TYPE patroni_pending_restart gauge") metrics.append("patroni_pending_restart{0} {1}" - .format(scope_label, int(patroni.postgresql.pending_restart))) + .format(labels, int(patroni.postgresql.pending_restart))) metrics.append("# HELP patroni_is_paused Value is 1 if auto failover is disabled, 0 otherwise.") metrics.append("# TYPE patroni_is_paused gauge") - metrics.append("patroni_is_paused{0} {1}".format(scope_label, int(postgres.get('pause', 0)))) + metrics.append("patroni_is_paused{0} {1}".format(labels, int(postgres.get('pause', 0)))) self.write_response(200, '\n'.join(metrics) + '\n', content_type='text/plain') From 0ab5b49757acea743be033a861dbd81a12e13304 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 5 Sep 2023 07:26:44 +0200 Subject: [PATCH 07/11] Introduce a dedicated postgres connection for REST API (#2833) Sharing a single connection between REST API and the main thread (doing heartbeats) was working mostly fine, except when Postgres becomes so slow that REST API queries start blocking the main loop. If the dedicated REST API connection isn't available we use the heartbeat connection as a fallback. --- patroni/api.py | 20 +++++++++++++++++--- tests/test_api.py | 42 ++++++++++++++++++++++++++++++++++-------- 2 files changed, 51 insertions(+), 11 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index a7a754a3..4647e7c2 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -1375,6 +1375,9 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): def query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: """Execute *sql* query with *params* and optionally return results. + .. note:: + Prefer to use own connection to postgres and fallback to ``heartbeat`` when own isn't available. + :param sql: the SQL statement to be run. :param params: positional arguments to be used as parameters for *sql*. @@ -1384,10 +1387,21 @@ 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. """ + # We first try to get a heartbeat connection because it is always required for the main thread. try: - return self.patroni.postgresql.query(sql, *params, retry=False) - except RetryFailedError as e: - raise PostgresConnectionException(str(e)) + heartbeat_connection = self.patroni.postgresql.connection_pool.get('heartbeat') + heartbeat_connection.get() # try to open psycopg connection to postgres + except psycopg.Error as exc: + raise PostgresConnectionException('connection problems') from exc + + try: + connection = self.patroni.postgresql.connection_pool.get('restapi') + connection.get() # try to open psycopg connection to postgres + except psycopg.Error: + logger.debug('restapi connection to postgres is not available') + connection = heartbeat_connection + + return connection.query(sql, *params) @staticmethod def _set_fd_cloexec(fd: socket.socket) -> None: diff --git a/tests/test_api.py b/tests/test_api.py index f433ca18..fa9a6280 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -11,9 +11,12 @@ from socketserver import ThreadingMixIn from patroni.api import RestApiHandler, RestApiServer from patroni.config import GlobalConfig from patroni.dcs import ClusterConfig, Member +from patroni.exceptions import PostgresConnectionException from patroni.ha import _MemberStatus +from patroni.psycopg import OperationalError from patroni.utils import RetryFailedError, tzutc +from . import MockConnect, psycopg_connect from .test_ha import get_cluster_initialized_without_leader @@ -21,8 +24,29 @@ future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5) postmaster_start_time = datetime.datetime.now(tzutc) -class MockPostgresql(object): +class MockConnection: + @staticmethod + def get(*args): + return psycopg_connect() + + @staticmethod + def query(sql, *params): + 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 MockConnectionPool: + + @staticmethod + def get(*args): + return MockConnection() + + +class MockPostgresql: + + connection_pool = MockConnectionPool() name = 'test' state = 'running' role = 'primary' @@ -54,12 +78,6 @@ 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 @@ -487,7 +505,7 @@ class TestRestApiHandler(unittest.TestCase): @patch('time.sleep', Mock()) def test_RestApiServer_query(self): - with patch.object(MockPostgresql, 'query', Mock(side_effect=RetryFailedError('bla'))): + with patch.object(MockConnection, 'query', Mock(side_effect=RetryFailedError('bla'))): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) @patch('time.sleep', Mock()) @@ -659,3 +677,11 @@ class TestRestApiServer(unittest.TestCase): def test_get_certificate_serial_number(self): self.assertIsNone(self.srv.get_certificate_serial_number()) + + def test_query(self): + with patch.object(MockConnection, 'get', Mock(side_effect=OperationalError)): + self.assertRaises(PostgresConnectionException, self.srv.query, 'SELECT 1') + with patch.object(MockConnection, 'get', Mock(side_effect=[MockConnect(), OperationalError])), \ + patch.object(MockConnection, 'query') as mock_query: + self.srv.query('SELECT 1') + mock_query.assert_called_once_with('SELECT 1') From 89a162e0008a0b8b2353d5cdcb3db6eff66f690b Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Tue, 5 Sep 2023 07:27:34 +0200 Subject: [PATCH 08/11] Return system id to the ctl list title (#2840) --- patroni/ctl.py | 8 +++++--- tests/test_ctl.py | 10 ++++++++++ 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index c8a396c9..3f49130e 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -1563,9 +1563,11 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, rows.append([member.get(n.lower().replace(' ', '_'), '') for n in columns]) title = 'Citus cluster' if is_citus_cluster else 'Cluster' - group_title = '' if group is None else 'group: {0}, '.format(group) - title_details = group_title and ' ({0}{1})'.format(group_title, initialize) - title = ' {0}: {1}{2} '.format(title, name, title_details) + title_details = f' ({initialize})' + if is_citus_cluster: + title_details = '' if group is None else f' (group: {group}, {initialize})' + + title = f' {title}: {name}{title_details} ' print_output(columns, rows, {'Group': 'r', 'Lag in MB': 'r', 'TL': 'r'}, fmt, title) if fmt not in ('pretty', 'topology'): # Omit service info when using machine-readable formats diff --git a/tests/test_ctl.py b/tests/test_ctl.py index f4f08296..b1468075 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -402,9 +402,19 @@ class TestCtl(unittest.TestCase): @patch('patroni.ctl.get_dcs') def test_members(self, mock_get_dcs): mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + result = self.runner.invoke(ctl, ['list']) assert '127.0.0.1' in result.output assert result.exit_code == 0 + assert 'Citus cluster: alpha -' in result.output + + result = self.runner.invoke(ctl, ['list', '--group', '0']) + assert 'Citus cluster: alpha (group: 0, 12345678901) -' in result.output + + with patch('patroni.ctl.load_config', Mock(return_value={'scope': 'alpha'})): + result = self.runner.invoke(ctl, ['list']) + assert 'Cluster: alpha (12345678901) -' in result.output + with patch('patroni.ctl.load_config', Mock(return_value={})): self.runner.invoke(ctl, ['list']) From 941e883ddedcd2a79382bc2ca2b98cdaa7310721 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 5 Sep 2023 07:41:45 +0200 Subject: [PATCH 09/11] Override write_leader_optime method in K8s implementation (#2850) It is being called when postgres is already shut down cleanly but there are no healthy replicas to take it over. Close https://github.com/zalando/patroni/issues/2837 Close https://github.com/zalando/patroni/pull/2838 --- patroni/dcs/kubernetes.py | 7 +++++++ tests/test_kubernetes.py | 4 ++++ 2 files changed, 11 insertions(+) diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 47ba4e7a..a88f4b23 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -1140,6 +1140,13 @@ class Kubernetes(AbstractDCS): """Unused""" raise NotImplementedError # pragma: no cover + def write_leader_optime(self, last_lsn: int) -> None: + """Write value for WAL LSN to ``optime`` annotation of the leader object. + + :param last_lsn: absolute WAL LSN in bytes. + """ + self.patch_or_create(self.leader_path, {self._OPTIME: str(last_lsn)}, patch=True, retry=False) + def _update_leader_with_retry(self, annotations: Dict[str, Any], resource_version: Optional[str], ips: List[str]) -> bool: retry = self._retry.copy() diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index c33bbecc..4f9f418c 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -436,6 +436,10 @@ class TestKubernetesEndpoints(BaseTestKubernetes): mock_logger_exception.assert_called_once() self.assertEqual(('create_config_service failed',), mock_logger_exception.call_args[0]) + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True) + def test_write_leader_optime(self): + self.k.write_leader_optime(12345) + def mock_watch(*args): return urllib3.HTTPResponse() From 30f0f132e835dc0d23b7cd06fa4f1bb3a3c1bad5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 Sep 2023 08:57:56 +0200 Subject: [PATCH 10/11] Don't start stopped postgres in pause (#2848) 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. Close https://github.com/zalando/patroni/issues/2834 --- patroni/postgresql/config.py | 34 ++++++++++++++++++++++++++++++---- tests/test_postgresql.py | 11 +++++++++++ 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 79141256..315bf8c7 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -14,7 +14,7 @@ from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Tuple from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value from ..collections import CaseInsensitiveDict, CaseInsensitiveSet from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name -from ..exceptions import PatroniFatalException +from ..exceptions import PatroniFatalException, PostgresConnectionException from ..file_perm import pg_perm from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath from ..validator import IntValidator, EnumValidator @@ -623,7 +623,24 @@ class ConfigHandler(object): 'recovery_target_action', 'standby_mode', self._triggerfile_wrong_name}) return CaseInsensitiveSet(self._RECOVERY_PARAMETERS - skip_params) - def _read_recovery_params(self) -> Tuple[Optional[CaseInsensitiveDict], Optional[bool]]: + def _read_recovery_params(self) -> Tuple[Optional[CaseInsensitiveDict], bool]: + """Read current recovery parameters values. + + .. note:: + We query Postgres only if we detected that Postgresql was restarted + or when at least one of the following files was updated: + + * ``postgresql.conf``; + * ``postgresql.auto.conf``; + * ``passfile`` that is used in the ``primary_conninfo``. + + :returns: a tuple with two elements: + + * :class:`CaseInsensitiveDict` object with current values of recovery parameters, + or ``None`` if no configuration files were updated; + + * ``True`` if new values of recovery parameters were queried, ``False`` otherwise. + """ if self._postgresql.is_starting(): return None, False @@ -644,11 +661,20 @@ class ConfigHandler(object): self._postgresql_conf_mtime = pg_conf_mtime self._auto_conf_mtime = auto_conf_mtime self._postmaster_ctime = postmaster_ctime - except Exception: + except Exception as exc: + if all((isinstance(exc, PostgresConnectionException), + self._postgresql_conf_mtime == pg_conf_mtime, + self._auto_conf_mtime == auto_conf_mtime, + self._passfile_mtime == passfile_mtime, + self._postmaster_ctime != postmaster_ctime)): + # We detected that the connection to postgres fails, but the process creation time of the postmaster + # doesn't match the old value. It is an indicator that Postgres crashed and either doing crash + # recovery or down. In this case we return values like nothing changed in the config. + return None, False values = None return values, True - def _read_recovery_params_pre_v12(self) -> Tuple[Optional[CaseInsensitiveDict], Optional[bool]]: + def _read_recovery_params_pre_v12(self) -> Tuple[Optional[CaseInsensitiveDict], bool]: recovery_conf_mtime = mtime(self._recovery_conf) passfile_mtime = mtime(self._passfile) if self._passfile else False if recovery_conf_mtime == self._recovery_conf_mtime and passfile_mtime == self._passfile_mtime: diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 29b25ee4..9ab3f52d 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -310,6 +310,17 @@ class TestPostgresql(BaseTestPostgresql): self.p.config.write_postgresql_conf() self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) + + # Config files changed, but can't connect to postgres + mock_get_pg_settings.side_effect = PostgresConnectionException('') + with patch('patroni.postgresql.config.mtime', mock_mtime): + self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) + + # Config files didn't change, but postgres crashed or in crash recovery + with patch.object(MockPostmaster, 'create_time', Mock(return_value=1234568), create=True): + self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) + + # Any other exception raised when executing the query mock_get_pg_settings.side_effect = Exception with patch('patroni.postgresql.config.mtime', mock_mtime): self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) From 19f20ec2ebb574aefe78d14eafd17f02555217c4 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 7 Sep 2023 12:56:07 +0200 Subject: [PATCH 11/11] Refactor replication slots handling (#2851) 1. make _get_members_slots() method return data in the same format as _get_permanent_slots() method 2. move conflicting name handling from get_replication_slots() to _get_members_slots() method 3. enrich structure returned by get_replication_slots() with the LSN of permanent logical slots reported by primary 4. use the added information in the SlotsHandler instead of fetching it from the Cluster.slots 5. bugfix: don't try to advance logical slot that doesn't match required configuration --- patroni/dcs/__init__.py | 68 +++++++++++++++++++++---------------- patroni/postgresql/slots.py | 21 ++++++------ tests/__init__.py | 2 +- tests/test_slots.py | 5 +-- 4 files changed, 52 insertions(+), 44 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index d28a59ce..4cef65ea 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -903,8 +903,16 @@ class Cluster(NamedTuple('Cluster', @property def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]: - """Dictionary of permanent replication slots.""" - return self.config and self.config.permanent_slots or {} + """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 + return ret @property def __permanent_physical_slots(self) -> Dict[str, Any]: @@ -929,7 +937,6 @@ class Cluster(NamedTuple('Cluster', Will log an error if: - * Conflicting slot names between members are found * Any logical slots are disabled, due to version compatibility, and *show_error* is ``True``. :param my_name: name of this node. @@ -942,21 +949,9 @@ class Cluster(NamedTuple('Cluster', :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) - - slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'} - for name in slot_members} - - if len(slots) < len(slot_members): - # Find which names are conflicting for a nicer error message - slot_conflicts: Dict[str, List[str]] = defaultdict(list) - for name in slot_members: - slot_conflicts[slot_name_from_member_name(name)].append(name) - logger.error("Following cluster members share a replication slot name: %s", - "; ".join(f"{', '.join(v)} map to {k}" - for k, v in slot_conflicts.items() if len(v) > 1)) - + 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) + disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( slots, permanent_slots, my_name, major_version) @@ -1016,7 +1011,7 @@ class Cluster(NamedTuple('Cluster', return disabled_permanent_logical_slots def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]: - """Get configured permanent slot names. + """Get configured permanent replication slots. .. note:: Permanent replication slots are only considered if ``use_slots`` configuration is enabled. @@ -1042,35 +1037,48 @@ class Cluster(NamedTuple('Cluster', 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. + 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. If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on the current primary, because that member would replicate from elsewhere. We still create the slot if the ``replicatefrom`` destination member is currently not a member of the cluster (fallback to the primary), or if ``replicatefrom`` destination member happens to be the current primary. + Will log an error if: + + * Conflicting slot names between members are found + :param my_name: name of this node. :param role: role of this node, if this is a ``primary`` or ``standby_leader`` return list of members replicating from this node. If not then return a list of members replicating as cascaded replicas from this node. - :returns: list of member names. + :returns: dictionary of physical replication slots that should exist on a given node. """ if not self.use_slots: - return [] + return {} + + # we always want to exclude the member with our name from the list + members = filter(lambda m: m.name != my_name, self.members) if role in ('master', 'primary', 'standby_leader'): - slot_members = [m.name for m in self.members - if m.name != my_name - and (m.replicatefrom is None - or m.replicatefrom == my_name - or not self.has_member(m.replicatefrom))] + members = [m for m in members if m.replicatefrom is None + or m.replicatefrom == my_name or not self.has_member(m.replicatefrom)] else: # only manage slots for replicas that replicate from this one, except for the leader among them - slot_members = [m.name for m in self.members - if m.replicatefrom == my_name and m.name != self.leader_name] - return slot_members + members = [m for m in members if m.replicatefrom == my_name and m.name != self.leader_name] + + slots = {slot_name_from_member_name(m.name): {'type': 'physical'} for m in members} + if len(slots) < len(members): + # Find which names are conflicting for a nicer error message + slot_conflicts: Dict[str, List[str]] = defaultdict(list) + for member in members: + slot_conflicts[slot_name_from_member_name(member.name)].append(member.name) + logger.error("Following cluster members share a replication slot name: %s", + "; ".join(f"{', '.join(v)} map to {k}" + 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: """Check if the given member node has permanent ``logical`` replication slots configured. diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 51a8fc5a..7391f543 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -434,7 +434,7 @@ class SlotsHandler: self._advance = SlotsAdvanceThread(self) return self._advance.schedule(slots) - def _ensure_logical_slots_replica(self, cluster: Cluster, slots: Dict[str, Any]) -> List[str]: + def _ensure_logical_slots_replica(self, slots: Dict[str, Any]) -> List[str]: """Update logical *slots* on replicas. If the logical slot already exists, copy state information into the replication slots structure stored in the @@ -444,7 +444,6 @@ class SlotsHandler: As logical slots can only be created when the primary is available, pass the list of slots that need to be copied back to the caller. They will be created on replicas with :meth:`SlotsHandler.copy_logical_slots`. - :param cluster: object containing stateful information for the cluster. :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot if the value is a dictionary with the key ``type`` and a value of ``logical``. @@ -459,15 +458,16 @@ class SlotsHandler: continue # If the logical already exists, copy some information about it into the original structure - if self._replication_slots.get(name, {}).get('datoid'): + if name in self._replication_slots and compare_slots(value, self._replication_slots[name]): self._copy_items(self._replication_slots[name], value) - if cluster.slots and name in cluster.slots: + if 'lsn' in value: # The slot has feedback in DCS try: # Skip slots that don't need to be advanced - if value['confirmed_flush_lsn'] < int(cluster.slots[name]): - advance_slots[value['database']][name] = int(cluster.slots[name]) + if value['confirmed_flush_lsn'] < int(value['lsn']): + advance_slots[value['database']][name] = int(value['lsn']) except Exception as e: - logger.error('Failed to parse "%s": %r', cluster.slots[name], e) - elif cluster.slots and name in cluster.slots: # We want to copy only slots with feedback in a DCS + logger.error('Failed to parse "%s": %r', value['lsn'], e) + elif name not in self._replication_slots and 'lsn' in value: + # 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, @@ -512,10 +512,9 @@ class SlotsHandler: if self._postgresql.is_primary(): self._logical_slots_processing_queue.clear() self._ensure_logical_slots_primary(slots) - elif cluster.slots and slots: + else: self.check_logical_slots_readiness(cluster, replicatefrom) - - ret = self._ensure_logical_slots_replica(cluster, slots) + ret = self._ensure_logical_slots_replica(slots) self._replication_slots = slots except Exception: diff --git a/tests/__init__.py b/tests/__init__.py index 79ef0229..68961af7 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -104,7 +104,7 @@ 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', 'a', 'b', 5, 100, 500)] + self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', '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()'): diff --git a/tests/test_slots.py b/tests/test_slots.py index add0fdf7..a35d465c 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -32,9 +32,9 @@ class TestSlotsHandler(BaseTestPostgresql): self.p._global_config = GlobalConfig({}) self.s = self.p.slots_handler self.p.start() - config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1) + config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}, 'ls2': None}}, 1) self.cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], - None, SyncState.empty(), None, {'ls': 12345}, None) + None, SyncState.empty(), None, {'ls': 12345, 'ls2': 12345}, None) def test_sync_replication_slots(self): config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'}, @@ -123,6 +123,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls']) self.cluster.slots['ls'] = 'a' self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) + self.cluster.config.data['slots']['ls']['database'] = 'b' with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])