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/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/citus.rst b/docs/citus.rst index 030cf93e..cb3f9a71 100644 --- a/docs/citus.rst +++ b/docs/citus.rst @@ -34,6 +34,8 @@ There are only a few simple rules you need to follow: After that you just need to start Patroni and it will handle the rest: +0. Patroni will set ``bootstrap.dcs.synchronous_mode`` to :ref:`quorum ` Patroni will @@ -73,36 +75,36 @@ It results in two major differences in ``patronictl`` behaviour when An example of ``patronictl list`` output for the Citus cluster:: postgres@coord1:~$ patronictl list demo - + Citus cluster: demo ----------+--------------+---------+----+-----------+ - | Group | Member | Host | Role | State | TL | Lag in MB | - +-------+---------+-------------+--------------+---------+----+-----------+ - | 0 | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | - | 0 | coord2 | 172.27.0.6 | Sync Standby | running | 1 | 0 | - | 0 | coord3 | 172.27.0.4 | Leader | running | 1 | | - | 1 | work1-1 | 172.27.0.8 | Sync Standby | running | 1 | 0 | - | 1 | work1-2 | 172.27.0.2 | Leader | running | 1 | | - | 2 | work2-1 | 172.27.0.5 | Sync Standby | running | 1 | 0 | - | 2 | work2-2 | 172.27.0.7 | Leader | running | 1 | | - +-------+---------+-------------+--------------+---------+----+-----------+ + + Citus cluster: demo ----------+----------------+---------+----+-----------+ + | Group | Member | Host | Role | State | TL | Lag in MB | + +-------+---------+-------------+----------------+---------+----+-----------+ + | 0 | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | + | 0 | coord2 | 172.27.0.6 | Quorum Standby | running | 1 | 0 | + | 0 | coord3 | 172.27.0.4 | Leader | running | 1 | | + | 1 | work1-1 | 172.27.0.8 | Quorum Standby | running | 1 | 0 | + | 1 | work1-2 | 172.27.0.2 | Leader | running | 1 | | + | 2 | work2-1 | 172.27.0.5 | Quorum Standby | running | 1 | 0 | + | 2 | work2-2 | 172.27.0.7 | Leader | running | 1 | | + +-------+---------+-------------+----------------+---------+----+-----------+ If we add the ``--group`` option, the output will change to:: postgres@coord1:~$ patronictl list demo --group 0 - + Citus cluster: demo (group: 0, 7179854923829112860) -----------+ - | Member | Host | Role | State | TL | Lag in MB | - +--------+-------------+--------------+---------+----+-----------+ - | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | - | coord2 | 172.27.0.6 | Sync Standby | running | 1 | 0 | - | coord3 | 172.27.0.4 | Leader | running | 1 | | - +--------+-------------+--------------+---------+----+-----------+ + + Citus cluster: demo (group: 0, 7179854923829112860) -+-----------+ + | Member | Host | Role | State | TL | Lag in MB | + +--------+-------------+----------------+---------+----+-----------+ + | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | + | coord2 | 172.27.0.6 | Quorum Standby | running | 1 | 0 | + | coord3 | 172.27.0.4 | Leader | running | 1 | | + +--------+-------------+----------------+---------+----+-----------+ postgres@coord1:~$ patronictl list demo --group 1 - + Citus cluster: demo (group: 1, 7179854923881963547) -----------+ - | Member | Host | Role | State | TL | Lag in MB | - +---------+------------+--------------+---------+----+-----------+ - | work1-1 | 172.27.0.8 | Sync Standby | running | 1 | 0 | - | work1-2 | 172.27.0.2 | Leader | running | 1 | | - +---------+------------+--------------+---------+----+-----------+ + + Citus cluster: demo (group: 1, 7179854923881963547) -+-----------+ + | Member | Host | Role | State | TL | Lag in MB | + +---------+------------+----------------+---------+----+-----------+ + | work1-1 | 172.27.0.8 | Quorum Standby | running | 1 | 0 | + | work1-2 | 172.27.0.2 | Leader | running | 1 | | + +---------+------------+----------------+---------+----+-----------+ Citus worker switchover ----------------------- @@ -118,28 +120,28 @@ new primary worker node is ready to accept read-write queries. An example of ``patronictl switchover`` on the worker cluster:: postgres@coord1:~$ patronictl switchover demo - + Citus cluster: demo ----------+--------------+---------+----+-----------+ - | Group | Member | Host | Role | State | TL | Lag in MB | - +-------+---------+-------------+--------------+---------+----+-----------+ - | 0 | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | - | 0 | coord2 | 172.27.0.6 | Sync Standby | running | 1 | 0 | - | 0 | coord3 | 172.27.0.4 | Leader | running | 1 | | - | 1 | work1-1 | 172.27.0.8 | Leader | running | 1 | | - | 1 | work1-2 | 172.27.0.2 | Sync Standby | running | 1 | 0 | - | 2 | work2-1 | 172.27.0.5 | Sync Standby | running | 1 | 0 | - | 2 | work2-2 | 172.27.0.7 | Leader | running | 1 | | - +-------+---------+-------------+--------------+---------+----+-----------+ + + Citus cluster: demo ----------+----------------+---------+----+-----------+ + | Group | Member | Host | Role | State | TL | Lag in MB | + +-------+---------+-------------+----------------+---------+----+-----------+ + | 0 | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | + | 0 | coord2 | 172.27.0.6 | Quorum Standby | running | 1 | 0 | + | 0 | coord3 | 172.27.0.4 | Leader | running | 1 | | + | 1 | work1-1 | 172.27.0.8 | Leader | running | 1 | | + | 1 | work1-2 | 172.27.0.2 | Quorum Standby | running | 1 | 0 | + | 2 | work2-1 | 172.27.0.5 | Quorum Standby | running | 1 | 0 | + | 2 | work2-2 | 172.27.0.7 | Leader | running | 1 | | + +-------+---------+-------------+----------------+---------+----+-----------+ Citus group: 2 Primary [work2-2]: Candidate ['work2-1'] []: When should the switchover take place (e.g. 2022-12-22T08:02 ) [now]: Current cluster topology - + Citus cluster: demo (group: 2, 7179854924063375386) -----------+ - | Member | Host | Role | State | TL | Lag in MB | - +---------+------------+--------------+---------+----+-----------+ - | work2-1 | 172.27.0.5 | Sync Standby | running | 1 | 0 | - | work2-2 | 172.27.0.7 | Leader | running | 1 | | - +---------+------------+--------------+---------+----+-----------+ + + Citus cluster: demo (group: 2, 7179854924063375386) -+-----------+ + | Member | Host | Role | State | TL | Lag in MB | + +---------+------------+----------------+---------+----+-----------+ + | work2-1 | 172.27.0.5 | Quorum Standby | running | 1 | 0 | + | work2-2 | 172.27.0.7 | Leader | running | 1 | | + +---------+------------+----------------+---------+----+-----------+ Are you sure you want to switchover cluster demo, demoting current primary work2-2? [y/N]: y 2022-12-22 07:02:40.33003 Successfully switched over to "work2-1" + Citus cluster: demo (group: 2, 7179854924063375386) ------+ @@ -150,17 +152,17 @@ An example of ``patronictl switchover`` on the worker cluster:: +---------+------------+---------+---------+----+-----------+ postgres@coord1:~$ patronictl list demo - + Citus cluster: demo ----------+--------------+---------+----+-----------+ - | Group | Member | Host | Role | State | TL | Lag in MB | - +-------+---------+-------------+--------------+---------+----+-----------+ - | 0 | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | - | 0 | coord2 | 172.27.0.6 | Sync Standby | running | 1 | 0 | - | 0 | coord3 | 172.27.0.4 | Leader | running | 1 | | - | 1 | work1-1 | 172.27.0.8 | Leader | running | 1 | | - | 1 | work1-2 | 172.27.0.2 | Sync Standby | running | 1 | 0 | - | 2 | work2-1 | 172.27.0.5 | Leader | running | 2 | | - | 2 | work2-2 | 172.27.0.7 | Sync Standby | running | 2 | 0 | - +-------+---------+-------------+--------------+---------+----+-----------+ + + Citus cluster: demo ----------+----------------+---------+----+-----------+ + | Group | Member | Host | Role | State | TL | Lag in MB | + +-------+---------+-------------+----------------+---------+----+-----------+ + | 0 | coord1 | 172.27.0.10 | Replica | running | 1 | 0 | + | 0 | coord2 | 172.27.0.6 | Quorum Standby | running | 1 | 0 | + | 0 | coord3 | 172.27.0.4 | Leader | running | 1 | | + | 1 | work1-1 | 172.27.0.8 | Leader | running | 1 | | + | 1 | work1-2 | 172.27.0.2 | Quorum Standby | running | 1 | 0 | + | 2 | work2-1 | 172.27.0.5 | Leader | running | 2 | | + | 2 | work2-2 | 172.27.0.7 | Quorum Standby | running | 2 | 0 | + +-------+---------+-------------+----------------+---------+----+-----------+ And this is how it looks on the coordinator side:: diff --git a/docs/rest_api.rst b/docs/rest_api.rst index 14e3aec2..c61cd03e 100644 --- a/docs/rest_api.rst +++ b/docs/rest_api.rst @@ -135,7 +135,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" } } @@ -182,7 +183,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" } } @@ -227,7 +229,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" } } @@ -271,7 +274,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" } } @@ -283,73 +287,73 @@ 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_quorum_standby Value is 1 if this node is a quorum standby replica, 0 otherwise. # TYPE patroni_quorum_standby gauge - patroni_quorum_standby{scope="batman"} 0 + patroni_quorum_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 @@ -388,6 +392,7 @@ Cluster status endpoints "lag": 0 } ], + "scope": "demo", "scheduled_switchover": { "at": "2023-09-24T10:36:00+02:00", "from": "patroni1", @@ -496,8 +501,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/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/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. diff --git a/patroni/api.py b/patroni/api.py index 1ac98696..90634428 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'] @@ -465,7 +469,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. @@ -543,117 +550,116 @@ 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, 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_quorum_standby Value is 1 if this node is a quorum standby, 0 otherwise.") metrics.append("# TYPE patroni_quorum_standby gauge") - metrics.append("patroni_quorum_standby{0} {1}".format(scope_label, int(postgres.get('quorum_standby', False)))) + metrics.append("patroni_quorum_standby{0} {1}".format(labels, int(postgres.get('quorum_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))) + metrics.append("patroni_pending_restart{0} {1}".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') @@ -1391,6 +1397,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*. @@ -1400,10 +1409,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/patroni/ctl.py b/patroni/ctl.py index b1d63d2b..3f49130e 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. @@ -1561,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/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 6758f25b..cf0a861a 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -915,8 +915,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]: @@ -941,7 +949,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. @@ -954,21 +961,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) @@ -1028,7 +1023,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. @@ -1054,35 +1049,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/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index c421abd3..ae6d3574 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': ''}) @@ -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() @@ -1269,13 +1276,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/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 668d2d11..cad0d6c0 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() @@ -282,7 +283,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 @@ -333,10 +334,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. @@ -699,7 +696,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.') @@ -770,7 +767,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..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: @@ -942,24 +968,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 +1006,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..7391f543 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 @@ -435,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 @@ -445,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``. @@ -460,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, @@ -513,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/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 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_api.py b/tests/test_api.py index 49150266..833afc13 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 @@ -490,7 +508,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()) @@ -662,3 +680,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') 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_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']) diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 2a1f6e26..d6aac8f0 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') @@ -434,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() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 510c2a90..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)) @@ -577,7 +588,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): 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'])