gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-26 07:30:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

Browse Source
This commit is contained in:
Alexander Kukushkin
2023-09-11 14:32:41 +02:00
parent 3a602f099e 19f20ec2eb
commit 1ea5d6bfb7
25 changed files with 458 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -1
View File
@@ -33,7 +33,7 @@ nosetests.xml
coverage.xml
htmlcov
junit.xml
features/output
features/output*
dummy
# Translations
docs/ENVIRONMENT.rst
+1 -1
View File
@@ -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.
docs/citus.rst
+54 -52
View File
@@ -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 <quorum_modes`
if it is not explicitly set to any other value.
1. ``citus`` extension will be automatically added to ``shared_preload_libraries``.
2. If ``max_prepared_transactions`` isn't explicitly set in the global
:ref:`dynamic configuration <dynamic_configuration>` 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::
docs/rest_api.rst
+34 -28
View File
@@ -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",
docs/yaml_configuration.rst
+1 -1
View File
@@ -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.
features/Dockerfile
+5 -2
View File
@@ -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.
patroni/api.py
+50 -30
View File
@@ -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:
patroni/ctl.py
+8 -4
View File
@@ -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
patroni/dcs/__init__.py
+38 -30
View File
@@ -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.
patroni/dcs/kubernetes.py
+10 -6
View File
@@ -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
patroni/postgresql/__init__.py
+6 -9
View File
@@ -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:
patroni/postgresql/bootstrap.py
+1 -1
View File
@@ -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
patroni/postgresql/citus.py
+6 -10
View File
@@ -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:
patroni/postgresql/config.py
+70 -23
View File
@@ -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 <patroni.postgresql.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 <patroni.postgresql.connection.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]
patroni/postgresql/connection.py
+84 -17
View File
@@ -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
patroni/postgresql/slots.py
+11 -13
View File
@@ -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:
requirements.docs.txt
+1 -2
View File
@@ -1,5 +1,4 @@
sphinx>=4
sphinx_rtd_theme>1
sphinxcontrib-apidoc
sphinx-github-style
pyyaml
sphinx-github-style<1.0.3
tests/__init__.py
+1 -1
View File
@@ -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()'):
tests/test_api.py
+34 -8
View File
@@ -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')
tests/test_bootstrap.py
+2 -2
View File
@@ -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'])
tests/test_citus.py
+1 -1
View File
@@ -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
tests/test_ctl.py
+10
View File
@@ -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'])
tests/test_kubernetes.py
+11 -5
View File
@@ -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()
tests/test_postgresql.py
+15 -1
View File
@@ -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):
tests/test_slots.py
+3 -2
View File
@@ -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'])