gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +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-10-23 15:28:54 +02:00
parent 5c6b34a757 d471f1156d
commit f32989124c
47 changed files with 592 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/install_deps.py
+2 -2
View File
@@ -45,8 +45,8 @@ def install_packages(what):
packages['exhibitor'] = packages['zookeeper']
packages = packages.get(what, [])
ver = versions.get(what)
if float(ver) == 15:
packages += ['postgresql-{0}-citus-12.0'.format(ver)]
if float(ver) >= 15:
packages += ['postgresql-{0}-citus-12.1'.format(ver)]
subprocess.call(['sudo', 'apt-get', 'update', '-y'])
return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev'] + packages)
.github/workflows/release.yaml
+4 -1
View File
@@ -24,8 +24,11 @@ jobs:
- name: Run tests and flake8
run: python .github/workflows/run_tests.py
- name: Install Python packaging build frontend
run: python -m pip install build
- name: Build a binary wheel and a source tarball
run: python setup.py sdist bdist_wheel
run: python -m build
- name: Publish distribution to Test PyPI
if: github.event_name == 'push'
README.rst
+11 -18
View File
@@ -77,23 +77,8 @@ There are a few options available:
sudo apt-get install python3-psycopg2 # install psycopg2 module on Debian/Ubuntu
sudo yum install python3-psycopg2 # install psycopg2 on RedHat/Fedora/CentOS
2. Install psycopg2 from the binary package
2. Specify one of `psycopg`, `psycopg2`, or `psycopg2-binary` in the list of dependencies when installing Patroni with pip (see below).
::
pip install psycopg2-binary
3. Install psycopg2 from source
::
pip install psycopg2>=2.5.4
4. Use psycopg 3.0 instead of psycopg2
::
pip install psycopg[binary]>=3.0.0
**General installation for pip**
@@ -119,12 +104,20 @@ raft
`pysyncobj` module in order to use python Raft implementation as DCS
aws
`boto3` in order to use AWS callbacks
all
all of the above (except psycopg family)
psycopg3
`psycopg[binary]>=3.0.0` module
psycopg2
`psycopg2>=2.5.4` module
psycopg2-binary
`psycopg2-binary` module
For example, the command in order to install Patroni together with dependencies for Etcd as a DCS and AWS callbacks is:
For example, the command in order to install Patroni together with psycopg3, dependencies for Etcd as a DCS, and AWS callbacks is:
::
pip install patroni[etcd,aws]
pip install patroni[psycopg3,etcd3,aws]
Note that external tools to call in the replica creation or custom bootstrap scripts (i.e. WAL-E) should be installed independently of Patroni.
docker/README.md
+1 -1
View File
@@ -78,7 +78,7 @@ Example session:
| demo | patroni3 | 172.22.0.4 | | running | 1 | 0 |
+---------+----------+------------+--------+---------+----+-----------+
postgres@patroni1:~$ etcdctl ls --recursive --sort -p /service/demo
postgres@patroni1:~$ etcdctl get --keys-only --prefix /service/demo
/service/demo/config
/service/demo/initialize
/service/demo/leader
docs/ENVIRONMENT.rst
-9
View File
@@ -24,15 +24,6 @@ Log
- **PATRONI\_LOG\_FILE\_SIZE**: Size of patroni.log file (in bytes) that triggers a log rolling.
- **PATRONI\_LOG\_LOGGERS**: Redefine logging level per python module. Example ``PATRONI_LOG_LOGGERS="{patroni.postmaster: WARNING, urllib3: DEBUG}"``
Bootstrap configuration
-----------------------
It is possible to create new database users right after the successful initialization of a new cluster. This process is defined by the following variables:
- **PATRONI\_<username>\_PASSWORD='<password>'**
- **PATRONI\_<username>\_OPTIONS='list,of,options'**
Example: defining ``PATRONI_admin_PASSWORD=strongpasswd`` and ``PATRONI_admin_OPTIONS='createrole,createdb'`` will cause creation of the user **admin** with the password **strongpasswd** that is allowed to create other users and databases.
Citus
-----
Enables integration Patroni with `Citus <https://docs.citusdata.com>`__. If configured, Patroni will take care of registering Citus worker nodes on the coordinator. You can find more information about Citus support :ref:`here <citus>`.
docs/citus.rst
+8 -4
View File
@@ -40,14 +40,18 @@ After that you just need to start Patroni and it will handle the rest:
2. If ``max_prepared_transactions`` isn't explicitly set in the global
:ref:`dynamic configuration <dynamic_configuration>` Patroni will
automatically set it to ``2*max_connections``.
3. The ``citus.database`` will be automatically created followed by ``CREATE EXTENSION citus``.
4. Current superuser :ref:`credentials <postgresql_settings>` will be added to the ``pg_dist_authinfo``
3. The ``citus.local_hostname`` GUC value will be adjusted from ``localhost`` to the
value that Patroni is using in order to connect to the local PostgreSQL
instance. The value sometimes should be different from the ``localhost``
because PostgreSQL might be not listening on it.
4. The ``citus.database`` will be automatically created followed by ``CREATE EXTENSION citus``.
5. Current superuser :ref:`credentials <postgresql_settings>` will be added to the ``pg_dist_authinfo``
table to allow cross-node communication. Don't forget to update them if
later you decide to change superuser username/password/sslcert/sslkey!
5. The coordinator primary node will automatically discover worker primary
6. The coordinator primary node will automatically discover worker primary
nodes and add them to the ``pg_dist_node`` table using the
``citus_add_node()`` function.
6. Patroni will also maintain ``pg_dist_node`` in case failover/switchover
7. Patroni will also maintain ``pg_dist_node`` in case failover/switchover
on the coordinator or worker clusters occurs.
patronictl
docs/contributing_guidelines.rst
+7
View File
@@ -43,6 +43,13 @@ After you have all dependencies installed, you can run the various test suites:
# Run the pytest suite in tests/:
python setup.py test
# Moreover, you may want to run tests in different scopes for debugging purposes,
# the -s option include print output during test execution.
# Tests in pytest typically follow the pattern: FILEPATH::CLASSNAME::TESTNAME.
pytest -s tests/test_api.py
pytest -s tests/test_api.py::TestRestApiHandler
pytest -s tests/test_api.py::TestRestApiHandler::test_do_GET
# Run the behave (https://behave.readthedocs.io/en/latest/) test suite in features/;
# modify DCS as desired (raft has no dependencies so is the easiest to start with):
DCS=raft python -m behave
docs/dynamic_configuration.rst
+6 -1
View File
@@ -57,7 +57,7 @@ In order to change the dynamic configuration you can use either :ref:`patronictl
- **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. With PostgreSQL 11 onwards permanent physical slots are created on all nodes and their position is advanced every **loop_wait** seconds. For PostgreSQL versions older than 11 permanent physical replication slots are maintained only on the current primary. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent replication slots requires **postgresql.use_slots** to be set to ``true``. If there are permanent logical replication slots defined Patroni will automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
- **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster <existing_data>` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni.
- **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current node it will not be created on this node. If you add a permanent physical replication slot which name matches the name of a Patroni member, Patroni will ensure that the slot that was created is not removed even if the corresponding member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when you want replication slots used by members to persist during temporary failures or when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster <existing_data>` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS, when the slot is no longer required, due to its effect on normal functioning of Patroni.
- **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``.
- **database**: the database name where logical slots should be created.
@@ -103,3 +103,8 @@ Note: if cluster topology is static (fixed number of nodes that never change the
node_name3:
type: physical
...
.. warning::
Permanent replication slots are synchronized only from the ``primary``/``standby_leader`` to replica nodes. That means, applications are supposed to be using them only from the leader node. Using them on replica nodes will cause indefinite growth of ``pg_wal`` on all other nodes in the cluster.
An exception to that rule are permanent physical slots that match the Patroni member names, if you happen to configure any. Those will be synchronized among all nodes as they are used for replication among them.
docs/installation.rst
+12 -17
View File
@@ -30,23 +30,10 @@ There are a few options available:
sudo apt-get install python3-psycopg2 # install psycopg2 module on Debian/Ubuntu
sudo yum install python3-psycopg2 # install psycopg2 on RedHat/Fedora/CentOS
2. Install psycopg2 from the binary package
2. Specify one of `psycopg`, `psycopg2`, or `psycopg2-binary` in the :ref:`list of dependencies <extras>` when installing Patroni with pip.
.. code-block:: shell
pip install psycopg2-binary
3. Install psycopg2 from source
.. code-block:: shell
pip install psycopg2>=2.5.4
4. Use psycopg 3.0 instead of psycopg2
.. code-block:: shell
pip install psycopg[binary]>=3.0.0
.. _extras:
General installation for pip
----------------------------
@@ -73,12 +60,20 @@ raft
`pysyncobj` module in order to use python Raft implementation as DCS
aws
`boto3` in order to use AWS callbacks
all
all of the above (except psycopg family)
psycopg
`psycopg[binary]>=3.0.0` module
psycopg2
`psycopg2>=2.5.4` module
psycopg2-binary
`psycopg2-binary` module
For example, the command in order to install Patroni together with dependencies for Etcd as a DCS and AWS callbacks is:
For example, the command in order to install Patroni together with psycopg3, dependencies for Etcd as a DCS, and AWS callbacks is:
.. code-block:: shell
pip install patroni[etcd,aws]
pip install patroni[psycopg3,etcd3,aws]
Note that external tools to call in the replica creation or custom bootstrap scripts (i.e. WAL-E) should be installed
independently of Patroni.
docs/patroni_configuration.rst
+9 -8
View File
@@ -70,15 +70,16 @@ There also are some parameters like **postgresql.listen**, **postgresql.data_dir
When applying the local or dynamic configuration options, the following actions are taken:
- The node first checks if there is a `postgresql.base.conf` or if the ``custom_conf`` parameter is set.
- If the ``custom_conf`` parameter is set, it will take the file specified on it as a base configuration, ignoring `postgresql.base.conf` and `postgresql.conf`.
- If the ``custom_conf`` parameter is not set and `postgresql.base.conf` exists, it contains the renamed "original" configuration and it will be used as a base configuration.
- If there is no ``custom_conf``` nor `postgresql.base.conf`, the original `postgresql.conf`` is taken and renamed to postgresql.base.conf.
- The dynamic options (with the exceptions above) are dumped into the `postgresql.conf`` and an include is set in
postgresql.conf to the used base configuration (either `postgresql.base.conf` or what is on ``custom_conf``). Therefore, we would be able to apply new options without re-reading the configuration file to check if the include is present not.
- The node first checks if there is a `postgresql.base.conf` file or if the ``custom_conf`` parameter is set.
- If the ``custom_conf`` parameter is set, the file it specifies is used as the base configuration, ignoring `postgresql.base.conf` and `postgresql.conf`.
- If the ``custom_conf`` parameter is not set and `postgresql.base.conf` exists, it contains the renamed "original" configuration and is used as the base configuration.
- If there is no ``custom_conf`` nor `postgresql.base.conf`, the original `postgresql.conf` is renamed to `postgresql.base.conf` and used as the base configuration.
- The dynamic options (with the exceptions above) are dumped into the `postgresql.conf` and an include is set in
`postgresql.conf` to the base configuration (either `postgresql.base.conf` or the file at ``custom_conf``).
Therefore, we would be able to apply new options without re-reading the configuration file to check if the include is present or not.
- Some parameters that are essential for Patroni to manage the cluster are overridden using the command line.
- If some of the options that require restart are changed (we should look at the context in pg_settings and at the actual
values of those options), a pending_restart flag of a given node is set. This flag is reset on any restart.
- If an option that requires restart is changed (we should look at the context in pg_settings and at the actual
values of those options), a pending_restart flag is set on that node. This flag is reset on any restart.
The parameters would be applied in the following order (run-time are given the highest priority):
docs/yaml_configuration.rst
-16
View File
@@ -49,24 +49,8 @@ Bootstrap configuration
- **- data-checksums**: Must be enabled when pg_rewind is needed on 9.3.
- **- encoding: UTF8**: default encoding for new databases.
- **- locale: UTF8**: default locale for new databases.
- **users**: Some additional users which need to be created after initializing new cluster, see :ref:`Bootstrap users configuration <bootstrap_users_configuration>` below.
- **post\_bootstrap** or **post\_init**: An additional script that will be executed after initializing the cluster. The script receives a connection string URL (with the cluster superuser as a user name). The PGPASSFILE variable is set to the location of pgpass file.
.. _bootstrap_users_configuration:
Bootstrap users configuration
=============================
Users which need to be created after initializing the cluster:
- **admin**: the name of user
- **password**: (optional) password for the user
- **options**: list of options for CREATE USER statement
- **- createrole**
- **- createdb**
.. _citus_settings:
Citus
features/citus.feature
+1 -1
View File
@@ -68,6 +68,6 @@ Feature: citus
And I receive a response output "+ttl: 20"
Then postgres4 is registered in the postgres2 as the primary in group 2 after 5 seconds
When I shut down postgres4
Then There is a transaction in progress on postgres0 changing pg_dist_node
Then there is a transaction in progress on postgres0 changing pg_dist_node after 5 seconds
When I run patronictl.py restart batman postgres2 --group 1 --force
Then a transaction finishes in 20 seconds
features/dcs_failsafe_mode.feature
+38 -6
View File
@@ -11,7 +11,7 @@ Feature: dcs failsafe mode
When I issue a GET request to http://127.0.0.1:8008/failsafe
Then I receive a response code 200
And I receive a response postgres0 http://127.0.0.1:8008/patroni
When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}}}
When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}},"slots":{"dcs_slot_1": null,"postgres0":null}}
Then I receive a response code 200
When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"dcs_slot_0": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
Then I receive a response code 200
@@ -44,14 +44,18 @@ Feature: dcs failsafe mode
@dcs-failsafe
@slot-advance
Scenario: check leader and replica are functioning while DCS is down
Given logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds
Given I get all changes from physical slot dcs_slot_1 on postgres0
Then physical slot dcs_slot_1 is in sync between postgres0 and postgres1 after 10 seconds
And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds
And DCS is down
Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds
Then postgres0 role is the primary after 10 seconds
And postgres1 role is the replica after 2 seconds
And replication works from postgres0 to postgres1 after 10 seconds
And I get all changes from logical slot dcs_slot_0 on postgres0
And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 seconds
When I get all changes from logical slot dcs_slot_0 on postgres0
And I get all changes from physical slot dcs_slot_1 on postgres0
Then logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 seconds
And physical slot dcs_slot_1 is in sync between postgres0 and postgres1 after 10 seconds
@dcs-failsafe
Scenario: check primary is demoted when one replica is shut down and DCS is down
@@ -70,15 +74,43 @@ Feature: dcs failsafe mode
And postgres1 role is the primary after 25 seconds
@dcs-failsafe
Scenario: check three-node cluster is functioning while DCS is down
Scenario: scale to three-node cluster
Given I start postgres0
And I start postgres2
Then "members/postgres2" key in DCS has state=running after 10 seconds
And "members/postgres0" key in DCS has state=running after 20 seconds
And Response on GET http://127.0.0.1:8008/failsafe contains postgres2 after 10 seconds
And replication works from postgres1 to postgres0 after 10 seconds
And replication works from postgres1 to postgres2 after 10 seconds
@dcs-failsafe
@slot-advance
Scenario: make sure permanent slots exist on replicas
Given I issue a PATCH request to http://127.0.0.1:8009/config with {"slots":{"dcs_slot_0":null,"dcs_slot_2":{"type":"logical","database":"postgres","plugin":"test_decoding"}}}
Then logical slot dcs_slot_2 is in sync between postgres1 and postgres0 after 20 seconds
And logical slot dcs_slot_2 is in sync between postgres1 and postgres2 after 20 seconds
When I get all changes from physical slot dcs_slot_1 on postgres1
Then physical slot dcs_slot_1 is in sync between postgres1 and postgres0 after 10 seconds
And physical slot dcs_slot_1 is in sync between postgres1 and postgres2 after 10 seconds
And physical slot postgres0 is in sync between postgres1 and postgres2 after 10 seconds
@dcs-failsafe
Scenario: check three-node cluster is functioning while DCS is down
Given DCS is down
Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds
Then Response on GET http://127.0.0.1:8009/primary contains failsafe_mode_is_active after 12 seconds
Then postgres1 role is the primary after 10 seconds
And postgres0 role is the replica after 2 seconds
And postgres2 role is the replica after 2 seconds
@dcs-failsafe
@slot-advance
Scenario: check that permanent slots are in sync between nodes while DCS is down
Given replication works from postgres1 to postgres0 after 10 seconds
And replication works from postgres1 to postgres2 after 10 seconds
When I get all changes from logical slot dcs_slot_2 on postgres1
And I get all changes from physical slot dcs_slot_1 on postgres1
Then logical slot dcs_slot_2 is in sync between postgres1 and postgres0 after 20 seconds
And logical slot dcs_slot_2 is in sync between postgres1 and postgres2 after 20 seconds
And physical slot dcs_slot_1 is in sync between postgres1 and postgres0 after 10 seconds
And physical slot dcs_slot_1 is in sync between postgres1 and postgres2 after 10 seconds
And physical slot postgres0 is in sync between postgres1 and postgres2 after 10 seconds
features/environment.py
+8 -3
View File
@@ -245,6 +245,10 @@ class PatroniController(AbstractController):
self.recursive_update(config, custom_config)
self.recursive_update(config, {
'log': {
'format': '%(asctime)s %(levelname)s [%(pathname)s:%(lineno)d - %(funcName)s]: %(message)s',
'loggers': {'patroni.postgresql.callback_executor': 'DEBUG'}
},
'bootstrap': {
'dcs': {
'loop_wait': 2,
@@ -650,9 +654,10 @@ class KubernetesController(AbstractExternalDcsController):
try:
if group is not None:
scope = '{0}-{1}'.format(scope, group)
ep = scope + {'leader': '', 'history': '-config', 'initialize': '-config'}.get(key, '-' + key)
rkey = 'leader' if key in ('status', 'failsafe') else key
ep = scope + {'leader': '', 'history': '-config', 'initialize': '-config'}.get(rkey, '-' + rkey)
e = self._api.read_namespaced_endpoints(ep, self._namespace)
if key != 'sync':
if key not in ('sync', 'status', 'failsafe'):
return e.metadata.annotations[key]
else:
return json.dumps(e.metadata.annotations)
@@ -688,7 +693,7 @@ class ZooKeeperController(AbstractExternalDcsController):
self._client = kazoo.client.KazooClient()
def process_name(self):
return "zookeeper"
return "java .*zookeeper"
def query(self, key, scope='batman', group=None):
import kazoo.exceptions
features/ignored_slots.feature
+1 -1
View File
@@ -50,7 +50,7 @@ Feature: ignored slots
And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds
And postgres1 does not have a logical replication slot named dummy_slot
And postgres1 does not have a replication slot named dummy_slot
# 3. After a failover the server (now a primary) still has the slot.
When I shut down postgres0
features/permanent_slots.feature
+48 -7
View File
@@ -3,32 +3,73 @@ Feature: permanent slots
Given I start postgres0
Then postgres0 is a leader after 10 seconds
And there is a non empty initialize key in DCS after 15 seconds
When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_physical": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}}
When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots":{"test_physical":0,"postgres0":0,"postgres1":0,"postgres3":0},"postgresql":{"parameters":{"wal_level":"logical"}}}
Then I receive a response code 200
And Response on GET http://127.0.0.1:8008/config contains slots after 10 seconds
When I start postgres1
And I start postgres2
And I configure and start postgres3 with a tag replicatefrom postgres2
Then postgres0 has a physical replication slot named test_physical after 10 seconds
And I start postgres1
And postgres0 has a physical replication slot named postgres1 after 10 seconds
And postgres0 has a physical replication slot named postgres2 after 10 seconds
And postgres2 has a physical replication slot named postgres3 after 10 seconds
@slot-advance
Scenario: check that logical permanent slots are created
Given I run patronictl.py restart batman postgres0 --force
And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots":{"test_logical":{"type":"logical","database":"postgres","plugin":"test_decoding"}}}
Then postgres0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
@slot-advance
Scenario: check that permanent slots are created on the replica
Scenario: check that permanent slots are created on replicas
Given postgres1 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
And Logical slot test_logical is in sync between postgres0 and postgres2 after 10 seconds
And Logical slot test_logical is in sync between postgres0 and postgres3 after 10 seconds
And postgres1 has a physical replication slot named test_physical after 2 seconds
And postgres2 has a physical replication slot named test_physical after 2 seconds
And postgres3 has a physical replication slot named test_physical after 2 seconds
@slot-advance
Scenario: check that permanent slots are advanced on the replica
Scenario: check permanent physical slots that match with member names
Given postgres0 has a physical replication slot named postgres3 after 2 seconds
And postgres1 has a physical replication slot named postgres0 after 2 seconds
And postgres1 has a physical replication slot named postgres3 after 2 seconds
And postgres2 has a physical replication slot named postgres0 after 2 seconds
And postgres2 has a physical replication slot named postgres3 after 2 seconds
And postgres2 has a physical replication slot named postgres1 after 2 seconds
And postgres1 does not have a replication slot named postgres2
And postgres3 does not have a replication slot named postgres2
@slot-advance
Scenario: check that permanent slots are advanced on replicas
Given I add the table replicate_me to postgres0
And I get all changes from physical slot test_physical on postgres0
When I get all changes from logical slot test_logical on postgres0
And I get all changes from physical slot test_physical on postgres0
Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
And Physical slot test_physical is in sync between postgres0 and postgres1 after 10 seconds
And Logical slot test_logical is in sync between postgres0 and postgres2 after 10 seconds
And Physical slot test_physical is in sync between postgres0 and postgres2 after 10 seconds
And Logical slot test_logical is in sync between postgres0 and postgres3 after 10 seconds
And Physical slot test_physical is in sync between postgres0 and postgres3 after 10 seconds
And Physical slot postgres1 is in sync between postgres0 and postgres2 after 10 seconds
And Physical slot postgres3 is in sync between postgres2 and postgres0 after 20 seconds
And Physical slot postgres3 is in sync between postgres2 and postgres1 after 10 seconds
And postgres1 does not have a replication slot named postgres2
And postgres3 does not have a replication slot named postgres2
@slot-advance
Scenario: check that only permanent slots are written to the /status key
Given "status" key in DCS has test_physical in slots
And "status" key in DCS has postgres0 in slots
And "status" key in DCS has postgres1 in slots
And "status" key in DCS does not have postgres2 in slots
And "status" key in DCS has postgres3 in slots
Scenario: check permanent physical replication slot after failover
Given I shut down postgres0
Given I shut down postgres3
And I shut down postgres2
And I shut down postgres0
Then postgres1 has a physical replication slot named test_physical after 10 seconds
And postgres1 has a physical replication slot named postgres0 after 10 seconds
And postgres1 has a physical replication slot named postgres3 after 10 seconds
features/recovery.feature
+2
View File
@@ -14,6 +14,8 @@ Feature: recovery
Then I receive a response code 200
And I receive a response role master
And I receive a response timeline 1
And "members/postgres0" key in DCS has state=running after 12 seconds
And replication works from postgres0 to postgres1 after 15 seconds
Scenario: check immediate failover when master_start_timeout=0
Given I issue a PATCH request to http://127.0.0.1:8008/config with {"master_start_timeout": 0}
features/standby_cluster.feature
+6 -9
View File
@@ -51,24 +51,21 @@ Feature: standby cluster
When I issue a GET request to http://127.0.0.1:8010/patroni
Then I receive a response code 200
And I receive a response replication_state streaming
And postgres1 does not have a logical replication slot named test_logical
And postgres1 does not have a replication slot named test_logical
Scenario: check switchover
When I run patronictl.py switchover batman1 --force
And I issue a GET request to http://127.0.0.1:8010/standby_leader
Then I receive a response code 200
And I receive a response role standby_leader
Given I run patronictl.py switchover batman1 --force
Then Status code on GET http://127.0.0.1:8010/standby_leader is 200 after 10 seconds
And postgres1 is replicating from postgres2 after 32 seconds
And there is a postgres2_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres2 data directory
Scenario: check failover
When I kill postgres2
And I kill postmaster on postgres2
Then postgres1 is replicating from postgres0 after 32 seconds
And Status code on GET http://127.0.0.1:8009/standby_leader is 200 after 10 seconds
When I issue a GET request to http://127.0.0.1:8009/primary
Then I receive a response code 503
And I sleep for 3 seconds
When I issue a GET request to http://127.0.0.1:8009/standby_leader
Then I receive a response code 200
And I receive a response role standby_leader
And replication works from postgres0 to postgres1 after 15 seconds
And there is a postgres1_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres1 data directory
And there is a postgres1_cb.log with "on_role_change replica batman1\non_role_change standby_leader batman1" in postgres1 data directory
features/steps/basic_replication.py
+1 -1
View File
@@ -110,7 +110,7 @@ def replication_works(context, primary, replica, time_limit):
context.execute_steps(u"""
When I add the table test_{0} to {1}
Then table test_{0} is present on {2} after {3} seconds
""".format(int(time()), primary, replica, time_limit))
""".format(str(time()).replace('.', '_').replace(',', '_'), primary, replica, time_limit))
@then('there is a "{message}" {level:w} in the {node} patroni log')
features/steps/citus.py
+12 -6
View File
@@ -115,12 +115,18 @@ def count_rows(context, name):
assert rows == context.insert_counter, "Distributed table doesn't have expected amount of rows"
@step("There is a transaction in progress on {name:w} changing pg_dist_node")
def check_transaction(context, name):
cur = context.pctl.query(name, "SELECT xact_start FROM pg_stat_activity WHERE pid <> pg_backend_pid()"
" AND state = 'idle in transaction' AND query ~ 'citus_update_node'")
assert cur.rowcount == 1, "There is no idle in transaction updating pg_dist_node"
context.xact_start = cur.fetchone()[0]
@step("there is a transaction in progress on {name:w} changing pg_dist_node after {time_limit:d} seconds")
def check_transaction(context, name, time_limit):
time_limit *= context.timeout_multiplier
max_time = time.time() + int(time_limit)
while time.time() < max_time:
cur = context.pctl.query(name, "SELECT xact_start FROM pg_stat_activity WHERE pid <> pg_backend_pid()"
" AND state = 'idle in transaction' AND query ~ 'citus_update_node'")
if cur.rowcount == 1:
context.xact_start = cur.fetchone()[0]
return
time.sleep(1)
assert False, f"There is no idle in transaction on {name} updating pg_dist_node after {time_limit} seconds"
@step("a transaction finishes in {timeout:d} seconds")
features/steps/slots.py
+16 -2
View File
@@ -1,3 +1,4 @@
import json
import time
from behave import step, then
@@ -36,8 +37,9 @@ def has_logical_replication_slot(context, pg_name, slot_name, plugin, time_limit
assert False, f"Error looking for slot {slot_name} on {pg_name} with plugin {plugin}"
@then('{pg_name:w} does not have a logical replication slot named {slot_name}')
def does_not_have_logical_replication_slot(context, pg_name, slot_name):
@step('{pg_name:w} does not have a replication slot named {slot_name:w}')
@then('{pg_name:w} does not have a replication slot named {slot_name:w}')
def does_not_have_replication_slot(context, pg_name, slot_name):
try:
row = context.pctl.query(pg_name, ("SELECT 1 FROM pg_replication_slots"
" WHERE slot_name = '{0}'").format(slot_name)).fetchone()
@@ -89,3 +91,15 @@ def has_physical_replication_slot(context, pg_name, slot_name, time_limit):
pass
time.sleep(1)
assert False, f"Physical slot {slot_name} doesn't exist after {time_limit} seconds"
@step('"{name}" key in DCS has {subkey:w} in {key:w}')
def dcs_key_contains(context, name, subkey, key):
response = json.loads(context.dcs_ctl.query(name))
assert key in response and subkey in response[key], f"{name} key in DCS doesn't have {subkey} in {key}"
@step('"{name}" key in DCS does not have {subkey:w} in {key:w}')
def dcs_key_does_not_contain(context, name, subkey, key):
response = json.loads(context.dcs_ctl.query(name))
assert key not in response or subkey not in response[key], f"{name} key in DCS has {subkey} in {key}"
patroni/__init__.py
+7 -53
View File
@@ -3,23 +3,14 @@
:var PATRONI_ENV_PREFIX: prefix for Patroni related configuration environment variables.
:var KUBERNETES_ENV_PREFIX: prefix for Kubernetes related configuration environment variables.
:var MIN_PSYCOPG2: minimum version of :mod:`psycopg2` required by Patroni to work.
:var MIN_PSYCOPG3: minimum version of :mod:`psycopg` required by Patroni to work.
"""
import sys
from typing import Any, Callable, Iterator, Tuple
from typing import Iterator, Tuple
PATRONI_ENV_PREFIX = 'PATRONI_'
KUBERNETES_ENV_PREFIX = 'KUBERNETES_'
MIN_PSYCOPG2 = (2, 5, 4)
def fatal(string: str, *args: Any) -> None:
"""Write a fatal message to stderr and exit with code ``1``.
:param string: message to be written before exiting.
"""
sys.exit('FATAL: ' + string.format(*args))
MIN_PSYCOPG3 = (3, 0, 0)
def parse_version(version: str) -> Tuple[int, ...]:
@@ -28,25 +19,25 @@ def parse_version(version: str) -> Tuple[int, ...]:
.. note::
Designed for easy comparison of software versions in Python.
:param version: human-readable software version, e.g. ``2.5.4``.
:param version: human-readable software version, e.g. ``2.5.4.dev1 (dt dec pq3 ext lo64)``.
:returns: tuple of *version* parts, each part as an integer.
:Example:
>>> parse_version('2.5.4')
>>> parse_version('2.5.4.dev1 (dt dec pq3 ext lo64)')
(2, 5, 4)
"""
def _parse_version(version: str) -> Iterator[int]:
"""Yield each part of a human-readable version string as an integer.
:param version: human-readable software version, e.g. ``2.5.4``.
:param version: human-readable software version, e.g. ``2.5.4.dev1``.
:yields: each part of *version* as an integer.
:Example:
>>> tuple(_parse_version('2.5.4'))
>>> tuple(_parse_version('2.5.4.dev1'))
(2, 5, 4)
"""
for e in version.split('.'):
@@ -55,40 +46,3 @@ def parse_version(version: str) -> Tuple[int, ...]:
except ValueError:
break
return tuple(_parse_version(version.split(' ')[0]))
def check_psycopg(_min_psycopg2: Tuple[int, ...] = MIN_PSYCOPG2,
_parse_version: Callable[[str], Tuple[int, ...]] = parse_version) -> None:
"""Ensure at least one among :mod:`psycopg2` or :mod:`psycopg` libraries are available in the environment.
.. note::
We pass ``MIN_PSYCOPG2`` and :func:`parse_version` as arguments to simplify usage of :func:`check_psycopg` from
the ``setup.py``.
.. note::
Patroni chooses :mod:`psycopg2` over :mod:`psycopg`, if possible.
If nothing meeting the requirements is found, then exit with a fatal message.
:param _min_psycopg2: minimum required version in case :mod:`psycopg2` is chosen.
:param _parse_version: function used to parse :mod:`psycopg2`/:mod:`psycopg` version into a comparable object.
"""
min_psycopg2_str = '.'.join(map(str, _min_psycopg2))
# try psycopg2
try:
from psycopg2 import __version__
if _parse_version(__version__) >= _min_psycopg2:
return
version_str = __version__.split(' ')[0]
except ImportError:
version_str = None
# try psycopg3
try:
from psycopg import __version__
except ImportError:
error = 'Patroni requires psycopg2>={0}, psycopg2-binary, or psycopg>=3.0'.format(min_psycopg2_str)
if version_str is not None:
error += ', but only psycopg2=={0} is available'.format(version_str)
fatal(error)
patroni/__main__.py
+42 -4
View File
@@ -10,8 +10,9 @@ import sys
import time
from argparse import Namespace
from typing import Any, Dict, Optional, TYPE_CHECKING
from typing import Any, Dict, List, Optional, TYPE_CHECKING
from patroni import MIN_PSYCOPG2, MIN_PSYCOPG3, parse_version
from patroni.daemon import AbstractPatroniDaemon, abstract_main, get_base_arg_parser
from patroni.tags import Tags
@@ -286,6 +287,45 @@ def process_arguments() -> Namespace:
return args
def check_psycopg() -> None:
"""Ensure at least one among :mod:`psycopg2` or :mod:`psycopg` libraries are available in the environment.
.. note::
Patroni chooses :mod:`psycopg2` over :mod:`psycopg`, if possible.
If nothing meeting the requirements is found, then exit with a fatal message.
"""
min_psycopg2_str = '.'.join(map(str, MIN_PSYCOPG2))
min_psycopg3_str = '.'.join(map(str, MIN_PSYCOPG3))
available_versions: List[str] = []
# try psycopg2
try:
from psycopg2 import __version__
if parse_version(__version__) >= MIN_PSYCOPG2:
return
available_versions.append('psycopg2=={0}'.format(__version__.split(' ')[0]))
except ImportError:
logger.debug('psycopg2 module is not available')
# try psycopg3
try:
from psycopg import __version__
if parse_version(__version__) >= MIN_PSYCOPG3:
return
available_versions.append('psycopg=={0}'.format(__version__.split(' ')[0]))
except ImportError:
logger.debug('psycopg module is not available')
error = f'FATAL: Patroni requires psycopg2>={min_psycopg2_str}, psycopg2-binary, or psycopg>={min_psycopg3_str}'
if available_versions:
error += ', but only {0} {1} available'.format(
' and '.join(available_versions),
'is' if len(available_versions) == 1 else 'are')
sys.exit(error)
def main() -> None:
"""Main entrypoint of :mod:`patroni.__main__`.
@@ -297,12 +337,10 @@ def main() -> None:
``patroni`` daemon as another process. In that case relevant signals received by the main process and forwarded
to ``patroni`` daemon process.
"""
from patroni import check_psycopg
check_psycopg()
args = process_arguments()
check_psycopg()
if os.getpid() != 1:
return patroni_main(args.configfile)
patroni/api.py
+3 -3
View File
@@ -467,7 +467,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
Write an HTTP response with JSON content based on the output of :func:`~patroni.utils.cluster_as_json`, with
HTTP status ``200`` and the JSON representation of the cluster topology.
"""
cluster = self.server.patroni.dcs.get_cluster(True)
cluster = self.server.patroni.dcs.get_cluster()
global_config = self.server.patroni.config.get_global_config(cluster)
response = cluster_as_json(cluster, global_config)
@@ -710,7 +710,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
"""
request = self._read_json_content()
if request:
cluster = self.server.patroni.dcs.get_cluster(True)
cluster = self.server.patroni.dcs.get_cluster()
if not (cluster.config and cluster.config.modify_version):
return self.send_error(503)
data = cluster.config.data.copy()
@@ -1187,7 +1187,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
patroni = self.server.patroni
if patroni.postgresql.citus_handler.is_coordinator() and patroni.ha.is_leader():
cluster = patroni.dcs.get_cluster(True)
cluster = patroni.dcs.get_cluster()
patroni.postgresql.citus_handler.handle_event(cluster, request)
self.write_response(200, 'OK')
patroni/dcs/__init__.py
+75 -40
View File
@@ -24,6 +24,7 @@ import dateutil.parser
from ..exceptions import PatroniFatalException
from ..utils import deep_compare, uri
from ..tags import Tags
from ..utils import parse_int
if TYPE_CHECKING: # pragma: no cover
from ..config import Config
@@ -158,7 +159,7 @@ def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS':
"""Attempt to load a Distributed Configuration Store from known available implementations.
.. note::
Using the list of available DCS modules returned by :func:`iter_dcs_modules` attempt to dynamically import and
Using the list of available DCS classes returned by :func:`iter_dcs_classes` attempt to dynamically
instantiate the class that implements a DCS using the abstract class :class:`AbstractDCS`.
Basic top-level configuration parameters retrieved from *config* are propagated to the DCS specific config
@@ -354,7 +355,7 @@ class Member(Tags, NamedTuple('Member',
@property
def lsn(self) -> Optional[int]:
"""Current LSN (receive/flush/replay)."""
return self.data.get('xlog_location')
return parse_int(self.data.get('xlog_location'))
class RemoteMember(Member):
@@ -986,29 +987,41 @@ class Cluster(NamedTuple('Cluster',
def is_physical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
"""Check whether provided configuration is for permanent physical replication slot.
:returns: ``True`` if this is a physical replication slot, otherwise ``False``.
:param value: configuration of the permanent replication slot.
:returns: ``True`` if *value* is a physical replication slot, otherwise ``False``.
"""
return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'
@staticmethod
def is_logical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
"""Check whether provided configuration is for permanent logical replication slot.
:param value: configuration of the permanent replication slot.
:returns: ``True`` if *value* is a logical replication slot, otherwise ``False``.
"""
return isinstance(value, dict) \
and value.get('type', 'logical') == 'logical' \
and bool(value.get('database') and value.get('plugin'))
@property
def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]:
"""Dictionary of permanent replication slots with their known LSN."""
leader = self.leader and self.leader.member
leader_name = slot_name_from_member_name(leader.name) if leader and leader.lsn else None
slots = self.slots or {}
ret: Dict[str, Union[Dict[str, Any], Any]] = deepcopy(self.config.permanent_slots if self.config else {})
members: Dict[str, int] = {slot_name_from_member_name(m.name): m.lsn or 0 for m in self.members}
slots: Dict[str, int] = {k: parse_int(v) or 0 for k, v in (self.slots or {}).items()}
for name, value in list(ret.items()):
if not value:
value = ret[name] = {}
if isinstance(value, dict):
if name in slots:
# If primary reported flush LSN for permanent slots we want to enrich our structure with it
value['lsn'] = slots[name]
elif self.is_physical_slot(value) and name == leader_name and leader and leader.lsn:
# there is no slot on the leader for itself, use `lsn` from the member key.
value['lsn'] = leader.lsn
# for permanent physical slots we want to get MAX LSN from the `Cluster.slots` and from the
# member with the matching name. It is necessary because we may have the replication slot on
# the primary that is streaming from the other standby node using the `replicatefrom` tag.
lsn = max(members.get(name, 0) if self.is_physical_slot(value) else 0, slots.get(name, 0))
if lsn:
value['lsn'] = lsn
else:
# Don't let anyone set 'lsn' in the global configuration :)
value.pop('lsn', None)
@@ -1022,8 +1035,7 @@ class Cluster(NamedTuple('Cluster',
@property
def __permanent_logical_slots(self) -> Dict[str, Any]:
"""Dictionary of permanent ``logical`` replication slots."""
return {name: value for name, value in self.__permanent_slots.items() if isinstance(value, dict)
and value.get('type', 'logical') == 'logical' and value.get('database') and value.get('plugin')}
return {name: value for name, value in self.__permanent_slots.items() if self.is_logical_slot(value)}
@property
def use_slots(self) -> bool:
@@ -1049,7 +1061,9 @@ class Cluster(NamedTuple('Cluster',
:returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
"""
slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role)
permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover, major_version)
permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster,
role=role, nofailover=nofailover,
major_version=major_version)
disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
slots, permanent_slots, my_name, major_version)
@@ -1060,8 +1074,7 @@ class Cluster(NamedTuple('Cluster',
return slots
@staticmethod
def _merge_permanent_slots(slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str,
def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str,
major_version: int) -> List[str]:
"""Merge replication *slots* for members with *permanent_slots*.
@@ -1096,7 +1109,7 @@ class Cluster(NamedTuple('Cluster',
slots[name] = value
continue
if value['type'] == 'logical' and value.get('database') and value.get('plugin'):
if self.is_logical_slot(value):
if major_version < SLOT_ADVANCE_AVAILABLE_VERSION:
disabled_permanent_logical_slots.append(name)
elif name in slots:
@@ -1109,7 +1122,7 @@ class Cluster(NamedTuple('Cluster',
logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
return disabled_permanent_logical_slots
def _get_permanent_slots(self, is_standby_cluster: bool, role: str,
def _get_permanent_slots(self, *, is_standby_cluster: bool, role: str,
nofailover: bool, major_version: int) -> Dict[str, Any]:
"""Get configured permanent replication slots.
@@ -1183,20 +1196,50 @@ class Cluster(NamedTuple('Cluster',
for k, v in slot_conflicts.items() if len(v) > 1))
return slots
def has_permanent_slots(self, my_name: str, nofailover: bool = False) -> bool:
def has_permanent_slots(self, my_name: str, *, is_standby_cluster: bool = False, nofailover: bool = False,
major_version: int = SLOT_ADVANCE_AVAILABLE_VERSION) -> bool:
"""Check if the given member node has permanent replication slots configured.
:param my_name: name of the member node to check.
:param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from
the outside because we want to protect from the ``/config`` key removal.
:param nofailover: ``True`` if this node is tagged to not be a failover candidate.
:param major_version: postgresql major version.
:returns: ``True`` if there are permanent replication slots configured, otherwise ``False``.
"""
members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, 'replica')
permanent_slots: Dict[str, Any] = self._get_permanent_slots(nofailover, 'replica', False,
SLOT_ADVANCE_AVAILABLE_VERSION)
role = 'replica'
members_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=is_standby_cluster,
role=role, nofailover=nofailover,
major_version=major_version)
slots = deepcopy(members_slots)
self._merge_permanent_slots(slots, permanent_slots, my_name, SLOT_ADVANCE_AVAILABLE_VERSION)
return len(slots) > len(members_slots)
self._merge_permanent_slots(slots, permanent_slots, my_name, major_version)
return len(slots) > len(members_slots) or any(self.is_physical_slot(v) for v in permanent_slots.values())
def filter_permanent_slots(self, slots: Dict[str, int], is_standby_cluster: bool,
major_version: int) -> Dict[str, int]:
"""Filter out all non-permanent slots from provided *slots* dict.
:param slots: slot names with LSN values
:param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from
the outside because we want to protect from the ``/config`` key removal.
:param major_version: postgresql major version.
:returns: a :class:`dict` object that contains only slots that are known to be permanent.
"""
if major_version < SLOT_ADVANCE_AVAILABLE_VERSION:
return {} # for legacy PostgreSQL we don't support permanent slots on standby nodes
permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster,
role='replica',
nofailover=False,
major_version=major_version)
members_slots = {slot_name_from_member_name(m.name) for m in self.members}
return {name: value for name, value in slots.items() if name in permanent_slots
and (self.is_physical_slot(permanent_slots[name])
or self.is_logical_slot(permanent_slots[name]) and name not in members_slots)}
def _has_permanent_logical_slots(self, my_name: str, nofailover: bool) -> bool:
"""Check if the given member node has permanent ``logical`` replication slots configured.
@@ -1560,9 +1603,6 @@ class AbstractDCS(abc.ABC):
primary and exception raised, instance would be demoted.
"""
def _bypass_caches(self) -> None:
"""Used only in Zookeeper."""
def __get_patroni_cluster(self, path: Optional[str] = None) -> Cluster:
"""Low level method to load a :class:`Cluster` object from DCS.
@@ -1605,15 +1645,14 @@ class AbstractDCS(abc.ABC):
dict.
"""
groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader)
if isinstance(groups, Cluster): # Zookeeper could return a cached version
cluster = groups
else:
cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
cluster.workers.update(groups)
if TYPE_CHECKING: # pragma: no cover
assert isinstance(groups, dict)
cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
cluster.workers.update(groups)
return cluster
def get_cluster(self, force: bool = False) -> Cluster:
"""Retrieve an appropriate cached or fresh view of DCS.
def get_cluster(self) -> Cluster:
"""Retrieve a fresh view of DCS.
.. note::
Stores copy of time, status and failsafe values for comparison in DCS update decisions.
@@ -1621,12 +1660,8 @@ class AbstractDCS(abc.ABC):
Returns either a Citus or Patroni implementation of :class:`Cluster` depending on availability.
:param force: a value of ``True`` will override Zookeeper caching features.
:returns:
"""
if force:
self._bypass_caches()
try:
cluster = self._get_citus_cluster() if self.is_citus_coordinator() else self.__get_patroni_cluster()
except Exception:
patroni/dcs/etcd3.py
+67 -39
View File
@@ -124,6 +124,10 @@ class AuthFailed(InvalidArgument):
error = "etcdserver: authentication failed, invalid user ID or password"
class AuthOldRevision(InvalidArgument):
error = "etcdserver: revision of auth store is old"
class PermissionDenied(Etcd3ClientError):
code = GRPCCode.PermissionDenied
error = "etcdserver: permission denied"
@@ -193,6 +197,12 @@ def build_range_request(key: str, range_end: Union[bytes, str, None] = None) ->
return fields
class ReAuthenticateMode(IntEnum):
NOT_REQUIRED = 0
REQUIRED = 1
WITHOUT_WATCHER_RESTART = 2
def _handle_auth_errors(func: Callable[..., Any]) -> Any:
def wrapper(self: 'Etcd3Client', *args: Any, **kwargs: Any) -> Any:
return self.handle_auth_errors(func, *args, **kwargs)
@@ -204,6 +214,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
ERROR_CLS = Etcd3Error
def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None:
self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
self._token = None
self._cluster_version: Tuple[int, ...] = tuple()
super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl)
@@ -282,7 +293,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
fields['retry'] = retry
return self.api_execute(self.version_prefix + method, self._MPOST, fields)
def authenticate(self) -> bool:
def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool:
if self._use_proxies and not self._cluster_version:
kwargs = self._prepare_common_parameters(1)
self._ensure_version_prefix(self._base_uri, **kwargs)
@@ -291,7 +302,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
logger.info('Trying to authenticate on Etcd...')
old_token, self._token = self._token, None
try:
response = self.call_rpc('/auth/authenticate', {'name': self.username, 'password': self.password})
response = self.call_rpc('/auth/authenticate', {'name': self.username, 'password': self.password}, retry)
except AuthNotEnabled:
logger.info('Etcd authentication is not enabled')
self._token = None
@@ -302,48 +313,65 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
self._token = response.get('token')
return old_token != self._token
def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
def retry(ex: Exception) -> Any:
if self.username and self.password:
self.authenticate()
return func(self, *args, **kwargs)
else:
logger.fatal('Username or password not set, authentication is not possible')
raise ex
def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any,
retry: Optional[Retry] = None, **kwargs: Any) -> Any:
exc = None
while True:
if self._reauthenticate_reason:
if self.username and self.password:
self.authenticate(
restart_watcher=self._reauthenticate_reason != ReAuthenticateMode.WITHOUT_WATCHER_RESTART,
retry=retry)
self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
if retry:
retry.ensure_deadline(0)
else:
msg = 'Username or password not set, authentication is not possible'
logger.fatal(msg)
raise exc or Etcd3Exception(msg)
try:
return func(self, *args, **kwargs)
except (UserEmpty, PermissionDenied) as e: # no token provided
# PermissionDenied is raised on 3.0 and 3.1
if self._cluster_version < (3, 3) and (not isinstance(e, PermissionDenied)
or self._cluster_version < (3, 2)):
raise UnsupportedEtcdVersion('Authentication is required by Etcd cluster but not '
'supported on version lower than 3.3.0. Cluster version: '
'{0}'.format('.'.join(map(str, self._cluster_version))))
return retry(e)
except InvalidAuthToken as e:
logger.error('Invalid auth token: %s', self._token)
return retry(e)
try:
return func(self, *args, retry=retry, **kwargs)
except (UserEmpty, PermissionDenied) as e: # no token provided
# PermissionDenied is raised on 3.0 and 3.1
if self._cluster_version < (3, 3) and (not isinstance(e, PermissionDenied)
or self._cluster_version < (3, 2)):
raise UnsupportedEtcdVersion('Authentication is required by Etcd cluster but not '
'supported on version lower than 3.3.0. Cluster version: '
'{0}'.format('.'.join(map(str, self._cluster_version))))
exc = e
except InvalidAuthToken as e:
logger.error('Invalid auth token: %s', self._token)
exc = e
except AuthOldRevision as e:
logger.error('Auth token is for old revision of auth store')
exc = e
self._reauthenticate_reason = ReAuthenticateMode.WITHOUT_WATCHER_RESTART \
if isinstance(exc, AuthOldRevision) else ReAuthenticateMode.REQUIRED
if not retry:
raise exc
retry.ensure_deadline(0.5, exc)
@_handle_auth_errors
def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True,
retry: Optional[Retry] = None) -> Dict[str, Any]:
*, retry: Optional[Retry] = None) -> Dict[str, Any]:
params = build_range_request(key, range_end)
params['serializable'] = serializable # For better performance. We can tolerate stale reads
return self.call_rpc('/kv/range', params, retry)
def prefix(self, key: str, serializable: bool = True, retry: Optional[Retry] = None) -> Dict[str, Any]:
return self.range(key, prefix_range_end(key), serializable, retry)
def prefix(self, key: str, serializable: bool = True, *, retry: Optional[Retry] = None) -> Dict[str, Any]:
return self.range(key, prefix_range_end(key), serializable, retry=retry)
@_handle_auth_errors
def lease_grant(self, ttl: int, retry: Optional[Retry] = None) -> str:
def lease_grant(self, ttl: int, *, retry: Optional[Retry] = None) -> str:
return self.call_rpc('/lease/grant', {'TTL': ttl}, retry)['ID']
def lease_keepalive(self, ID: str, retry: Optional[Retry] = None) -> Optional[str]:
def lease_keepalive(self, ID: str, *, retry: Optional[Retry] = None) -> Optional[str]:
return self.call_rpc('/lease/keepalive', {'ID': ID}, retry).get('result', {}).get('TTL')
@_handle_auth_errors
def txn(self, compare: Dict[str, Any], success: Dict[str, Any],
failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
failure: Optional[Dict[str, Any]] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]:
fields = {'compare': [compare], 'success': [success]}
if failure:
fields['failure'] = [failure]
@@ -352,7 +380,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
@_handle_auth_errors
def put(self, key: str, value: str, lease: Optional[str] = None, create_revision: Optional[str] = None,
mod_revision: Optional[str] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
mod_revision: Optional[str] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]:
fields = {'key': base64_encode(key), 'value': base64_encode(value)}
if lease:
fields['lease'] = lease
@@ -367,14 +395,14 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
@_handle_auth_errors
def deleterange(self, key: str, range_end: Union[bytes, str, None] = None,
mod_revision: Optional[str] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
mod_revision: Optional[str] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]:
fields = build_range_request(key, range_end)
if mod_revision is None:
return self.call_rpc('/kv/deleterange', fields, retry)
compare = {'target': 'MOD', 'mod_revision': mod_revision, 'key': fields['key']}
return self.txn(compare, {'request_delete_range': fields}, retry=retry)
def deleteprefix(self, key: str, retry: Optional[Retry] = None) -> Dict[str, Any]:
def deleteprefix(self, key: str, *, retry: Optional[Retry] = None) -> Dict[str, Any]:
return self.deleterange(key, prefix_range_end(key), retry=retry)
def watchrange(self, key: str, range_end: Union[bytes, str, None] = None,
@@ -574,9 +602,9 @@ class PatroniEtcd3Client(Etcd3Client):
super(PatroniEtcd3Client, self).set_base_uri(value)
self._restart_watcher()
def authenticate(self) -> bool:
ret = super(PatroniEtcd3Client, self).authenticate()
if ret:
def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool:
ret = super(PatroniEtcd3Client, self).authenticate(restart_watcher=restart_watcher, retry=retry)
if ret and restart_watcher:
self._restart_watcher()
return ret
@@ -631,8 +659,8 @@ class PatroniEtcd3Client(Etcd3Client):
return ret
def txn(self, compare: Dict[str, Any], success: Dict[str, Any],
failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry)
failure: Optional[Dict[str, Any]] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]:
ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry=retry)
# Here we abuse the fact that the `failure` is only set in the call from update_leader().
# In all other cases the txn() call failure may be an indicator of a stale cache,
# and therefore we want to restart watcher.
@@ -676,12 +704,12 @@ class Etcd3(AbstractEtcd):
if not force and self._lease and self._last_lease_refresh + self._loop_wait > time.time():
return False
if self._lease and not self._client.lease_keepalive(self._lease, retry):
if self._lease and not self._client.lease_keepalive(self._lease, retry=retry):
self._lease = None
ret = not self._lease
if ret:
self._lease = self._client.lease_grant(self._ttl, retry)
self._lease = self._client.lease_grant(self._ttl, retry=retry)
self._last_lease_refresh = time.time()
return ret
patroni/dcs/zookeeper.py
+31 -85
View File
@@ -116,10 +116,7 @@ class ZooKeeper(AbstractDCS):
timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
sleep_func=time.sleep), command_retry=KazooRetry(max_delay=1, max_tries=-1,
deadline=config['retry_timeout'], sleep_func=time.sleep), **kwargs)
self._client.add_listener(self.session_listener)
self._fetch_cluster: bool = True
self._fetch_status: bool = True
self.__last_member_data: Optional[Dict[str, Any]] = None
self._orig_kazoo_connect = self._client._connection._connect
@@ -142,18 +139,9 @@ class ZooKeeper(AbstractDCS):
ret = self._orig_kazoo_connect(*args)
return max(self.loop_wait - 2, 2) * 1000, ret[1]
def session_listener(self, state: str) -> None:
if state in [KazooState.SUSPENDED, KazooState.LOST]:
self.cluster_watcher(None)
def status_watcher(self, event: Optional[WatchedEvent]) -> None:
self._fetch_status = True
self.event.set()
def cluster_watcher(self, event: Optional[WatchedEvent]) -> None:
self._fetch_cluster = True
if not event or event.state != KazooState.CONNECTED or event.path.startswith(self.client_path('')):
self.status_watcher(event)
def _watcher(self, event: WatchedEvent) -> None:
if event.state != KazooState.CONNECTED or event.path.startswith(self.client_path('')):
self.event.set()
def reload_config(self, config: Union['Config', Dict[str, Any]]) -> None:
self.set_retry_timeout(config['retry_timeout'])
@@ -202,75 +190,66 @@ class ZooKeeper(AbstractDCS):
return None
def get_status(self, path: str, leader: Optional[Leader]) -> Status:
watch = self.status_watcher if not leader or leader.name != self._name else None
status = self.get_node(path + self._STATUS, watch)
status = self.get_node(path + self._STATUS)
if not status:
status = self.get_node(path + self._LEADER_OPTIME, watch)
if status:
self._fetch_status = False
status = self.get_node(path + self._LEADER_OPTIME)
return Status.from_node(status and status[0])
@staticmethod
def member(name: str, value: str, znode: ZnodeStat) -> Member:
return Member.from_node(znode.version, name, znode.ephemeralOwner, value)
def get_children(self, key: str, watch: Optional[Callable[[WatchedEvent], None]] = None) -> List[str]:
def get_children(self, key: str) -> List[str]:
try:
return self._client.get_children(key, watch)
return self._client.get_children(key)
except NoNodeError:
return []
def load_members(self, path: str) -> List[Member]:
members: List[Member] = []
for member in self.get_children(path + self._MEMBERS, self.cluster_watcher):
for member in self.get_children(path + self._MEMBERS):
data = self.get_node(path + self._MEMBERS + member)
if data is not None:
members.append(self.member(member, *data))
return members
def _cluster_loader(self, path: str) -> Cluster:
self._fetch_cluster = False
self.event.clear()
nodes = set(self.get_children(path, self.cluster_watcher))
if not nodes:
self._fetch_cluster = True
nodes = set(self.get_children(path))
# get initialize flag
initialize = (self.get_node(path + self._INITIALIZE) or [None])[0] if self._INITIALIZE in nodes else None
# get global dynamic configuration
config = self.get_node(path + self._CONFIG, watch=self.cluster_watcher) if self._CONFIG in nodes else None
config = self.get_node(path + self._CONFIG, watch=self._watcher) if self._CONFIG in nodes else None
config = config and ClusterConfig.from_node(config[1].version, config[0], config[1].mzxid)
# get timeline history
history = self.get_node(path + self._HISTORY, watch=self.cluster_watcher) if self._HISTORY in nodes else None
history = self.get_node(path + self._HISTORY) if self._HISTORY in nodes else None
history = history and TimelineHistory.from_node(history[1].mzxid, history[0])
# get synchronization state
sync = self.get_node(path + self._SYNC, watch=self.cluster_watcher) if self._SYNC in nodes else None
sync = self.get_node(path + self._SYNC) if self._SYNC in nodes else None
sync = SyncState.from_node(sync and sync[1].version, sync and sync[0])
# get list of members
members = self.load_members(path) if self._MEMBERS[:-1] in nodes else []
# get leader
leader = self.get_node(path + self._LEADER) if self._LEADER in nodes else None
leader = self.get_node(path + self._LEADER, watch=self._watcher) if self._LEADER in nodes else None
if leader:
member = Member(-1, leader[0], None, {})
member = ([m for m in members if m.name == leader[0]] or [member])[0]
leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
self._fetch_cluster = member.version == -1
# get last known leader lsn and slots
status = self.get_status(path, leader)
# failover key
failover = self.get_node(path + self._FAILOVER, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
failover = self.get_node(path + self._FAILOVER) if self._FAILOVER in nodes else None
failover = failover and Failover.from_node(failover[1].version, failover[0])
# get failsafe topology
failsafe = self.get_node(path + self._FAILSAFE, watch=self.cluster_watcher) if self._FAILSAFE in nodes else None
failsafe = self.get_node(path + self._FAILSAFE) if self._FAILSAFE in nodes else None
try:
failsafe = json.loads(failsafe[0]) if failsafe else None
except Exception:
@@ -279,45 +258,20 @@ class ZooKeeper(AbstractDCS):
return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe)
def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
fetch_cluster = False
ret: Dict[int, Cluster] = {}
for node in self.get_children(path, self.cluster_watcher):
for node in self.get_children(path):
if citus_group_re.match(node):
ret[int(node)] = self._cluster_loader(path + node + '/')
fetch_cluster = fetch_cluster or self._fetch_cluster
self._fetch_cluster = fetch_cluster
return ret
def _load_cluster(
self, path: str, loader: Callable[[str], Union[Cluster, Dict[int, Cluster]]]
) -> Union[Cluster, Dict[int, Cluster]]:
cluster = self.cluster if path == self._base_path + '/' else None
if self._fetch_cluster or cluster is None:
try:
cluster = self._client.retry(loader, path)
except Exception:
logger.exception('get_cluster')
self.cluster_watcher(None)
raise ZooKeeperError('ZooKeeper in not responding properly')
# The /status ZNode was updated or doesn't exist
elif self._fetch_status and not self._fetch_cluster or not cluster.last_lsn \
or cluster.has_permanent_slots(self._name) and not cluster.slots:
# If current node is the leader just clear the event without fetching anything (we are updating the /status)
if cluster.leader and cluster.leader.name == self._name:
self.event.clear()
else:
try:
status = self.get_status(self.client_path(''), cluster.leader)
self.event.clear()
new_cluster: List[Any] = list(cluster)
new_cluster[3] = status
cluster = Cluster(*new_cluster)
except Exception:
pass
return cluster
def _bypass_caches(self) -> None:
self._fetch_cluster = True
try:
return self._client.retry(loader, path)
except Exception:
logger.exception('get_cluster')
raise ZooKeeperError('ZooKeeper in not responding properly')
def _create(self, path: str, value: bytes, retry: bool = False, ephemeral: bool = False) -> bool:
try:
@@ -380,20 +334,9 @@ class ZooKeeper(AbstractDCS):
member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
member_data = self.__last_member_data or member and member.data
if member and member_data:
is_leader = data.get('role') in ('master', 'primary', 'standby_leader')
checkpoint_after_promote_changed = member_data.get('checkpoint_after_promote') \
!= data.get('checkpoint_after_promote')
state_running_changed = member_data.get('state') != data.get('state') \
and 'running' in (member_data.get('state'), data.get('state'))
tags_changed = not deep_compare(member_data.get('tags', {}), data.get('tags', {}))
# We want delete the member ZNode if:
# - our session doesn't match with session id on our member key; or
# - we want to notify leader if some important fields in the member key changed; or
# - if we are the leader and want to notify replicas about checkpoint_after_promote;
if self._client.client_id is not None and member.session != self._client.client_id[0] \
or is_leader and checkpoint_after_promote_changed \
or not is_leader and (state_running_changed or tags_changed):
# We want delete the member ZNode if our session doesn't match with session id on our member key
if self._client.client_id is not None and member.session != self._client.client_id[0]:
logger.warning('Recreating the member ZNode due to ownership mismatch')
try:
self._client.delete_async(self.member_path).get(timeout=1)
except NoNodeError:
@@ -492,7 +435,10 @@ class ZooKeeper(AbstractDCS):
return self.set_sync_state_value("{}", version) is not False
def watch(self, leader_version: Optional[int], timeout: float) -> bool:
ret = super(ZooKeeper, self).watch(leader_version, timeout + 0.5)
if ret and not self._fetch_status:
self._fetch_cluster = True
return ret or self._fetch_cluster
if leader_version:
timeout += 0.5
try:
return super(ZooKeeper, self).watch(leader_version, timeout)
finally:
self.event.clear()
patroni/ha.py
+51 -6
View File
@@ -14,7 +14,7 @@ from . import psycopg
from .__main__ import Patroni
from .async_executor import AsyncExecutor, CriticalTask
from .collections import CaseInsensitiveSet
from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status, SyncState
from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status, SyncState, slot_name_from_member_name
from .exceptions import DCSError, PostgresConnectionException, PatroniFatalException
from .postgresql.callback_executor import CallbackAction
from .postgresql.misc import postgres_version_to_int
@@ -283,12 +283,32 @@ class Ha(object):
ret[self.state_handler.name] = self.patroni.api.connection_string
return ret
def update_lock(self, write_leader_optime: bool = False) -> bool:
def update_lock(self, update_status: bool = False) -> bool:
"""Update the leader lock in DCS.
.. note::
After successful update of the leader key the :meth:`AbstractDCS.update_leader` method could also
optionally update the ``/status`` and ``/failsafe`` keys.
The ``/status`` key contains the last known LSN on the leader node and the last known state
of permanent replication slots including permanent physical replication slot for the leader.
Last, but not least, this method calls a :meth:`Watchdog.keepalive` method after the leader key
was successfully updated.
:param update_status: ``True`` if we also need to update the ``/status`` key in DCS, otherwise ``False``.
:returns: ``True`` if the leader key was successfully updated and we can continue to run postgres
as a ``primary`` or as a ``standby_leader``, otherwise ``False``.
"""
last_lsn = slots = None
if write_leader_optime:
if update_status:
try:
last_lsn = self.state_handler.last_operation()
slots = self.state_handler.slots()
slots = self.cluster.filter_permanent_slots(
{**self.state_handler.slots(), slot_name_from_member_name(self.state_handler.name): last_lsn},
self.is_standby_cluster(),
self.state_handler.major_version)
except Exception:
logger.exception('Exception when called state_handler.last_operation()')
if TYPE_CHECKING: # pragma: no cover
@@ -595,7 +615,9 @@ class Ha(object):
"""
# The standby leader or when there is no standby leader we want to follow
# the remote member, except when there is no standby leader in pause.
if self.is_standby_cluster() and (self.has_lock(False) or self.cluster.is_unlocked() and not self.is_paused()):
if self.is_standby_cluster() \
and (cluster.leader and cluster.leader.name and cluster.leader.name == self.state_handler.name
or cluster.is_unlocked() and not self.is_paused()):
node_to_follow = self.get_remote_member()
# If replicatefrom tag is set, try to follow the node mentioned there, otherwise, follow the leader.
elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
@@ -1049,6 +1071,26 @@ class Ha(object):
return False
def check_failsafe_topology(self) -> bool:
"""Check whether we could continue to run as a primary by calling all members from the failsafe topology.
.. note::
If the ``/failsafe`` key contains invalid data or if the ``name`` of our node is missing in
the ``/failsafe`` key, we immediately give up and return ``False``.
We send the JSON document in the POST request with the following fields:
* ``name`` - the name of our node;
* ``conn_url`` - connection URL to the postgres, which is reachable from other nodes;
* ``api_url`` - connection URL to Patroni REST API on this node reachable from other nodes;
* ``slots`` - a :class:`dict` with replication slots that exist on the leader node, including the primary
itself with the last known LSN, because there could be a permanent physical slot on standby nodes.
Standby nodes are using information from the ``slots`` dict to advance position of permanent
replication slots while DCS is not accessible in order to avoid indefinite growth of ``pg_wal``.
:returns: ``True`` if all members from the ``/failsafe`` topology agree that this node could continue to
run as a ``primary``, or ``False`` if some of standby nodes are not accessible or don't agree.
"""
failsafe = self.dcs.failsafe
if not isinstance(failsafe, dict) or self.state_handler.name not in failsafe:
return False
@@ -1058,7 +1100,10 @@ class Ha(object):
'api_url': self.patroni.api.connection_string,
}
try:
data['slots'] = self.state_handler.slots()
data['slots'] = {
**self.state_handler.slots(),
slot_name_from_member_name(self.state_handler.name): self.state_handler.last_operation()
}
except Exception:
logger.exception('Exception when called state_handler.slots()')
members = [RemoteMember(name, {'api_url': url})
patroni/postgresql/__init__.py
+11 -5
View File
@@ -456,15 +456,21 @@ class Postgresql(object):
return
if self._global_config.is_standby_cluster:
self._has_permanent_slots = False
# Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback
self.set_enforce_hot_standby_feedback(False)
elif cluster and cluster.config and cluster.config.modify_version:
self._has_permanent_slots = cluster.has_permanent_slots(self.name, nofailover)
if cluster and cluster.config and cluster.config.modify_version:
# We want to enable hot_standby_feedback if the replica is supposed
# to have a logical slot or in case if it is the cascading replica.
self.set_enforce_hot_standby_feedback(
self.can_advance_slots and cluster.should_enforce_hot_standby_feedback(self.name, nofailover))
self.set_enforce_hot_standby_feedback(not self._global_config.is_standby_cluster and self.can_advance_slots
and cluster.should_enforce_hot_standby_feedback(self.name,
nofailover))
self._has_permanent_slots = cluster.has_permanent_slots(
my_name=self.name,
is_standby_cluster=self._global_config.is_standby_cluster,
nofailover=nofailover,
major_version=self.major_version)
def _cluster_info_state_get(self, name: str) -> Optional[Any]:
if not self._cluster_info_state:
patroni/postgresql/bootstrap.py
+3
View File
@@ -400,6 +400,9 @@ BEGIN
END;$$""".format(f, quote_ident(rewind['username'], postgresql.connection()))
postgresql.query(sql)
if config.get('users'):
logger.warning('User creation via "bootstrap.users" will be removed in v4.0.0')
for name, value in (config.get('users') or {}).items():
if all(name != a.get('username') for a in (superuser, replication, rewind)):
self.create_or_update_role(name, value.get('password'), value.get('options', []))
patroni/postgresql/callback_executor.py
+4 -1
View File
@@ -1,8 +1,9 @@
import logging
import sys
from enum import Enum
from threading import Condition, Thread
from typing import List
from typing import Any, Dict, List
from .cancellable import CancellableExecutor, CancellableSubprocess
@@ -53,6 +54,8 @@ class CallbackExecutor(CancellableExecutor, Thread):
If it couldn't be killed we wait until it finishes.
:param cmd: command to be executed"""
kwargs: Dict[str, Any] = {'stacklevel': 3} if sys.version_info >= (3, 8) else {}
logger.debug('CallbackExecutor.call(%s)', cmd, **kwargs)
if cmd[-3] == CallbackAction.ON_RELOAD:
return self._on_reload_executor.call_nowait(cmd)
patroni/postgresql/citus.py
+3
View File
@@ -403,6 +403,9 @@ class CitusHandler(Thread):
# Resharding in Citus implemented using logical replication
parameters['wal_level'] = 'logical'
# Sometimes Citus needs to connect to the local postgres. We will do it the same way as Patroni does.
parameters['citus.local_hostname'] = self._postgresql.connection_pool.conn_kwargs.get('host', 'localhost')
def ignore_replication_slot(self, slot: Dict[str, str]) -> bool:
if isinstance(self._config, dict) and self._postgresql.is_primary() and\
slot['type'] == 'logical' and slot['database'] == self._config['database']:
patroni/postgresql/slots.py
+5 -10
View File
@@ -16,7 +16,6 @@ from .misc import format_lsn, fsync_dir
from ..dcs import Cluster, Leader
from ..file_perm import pg_perm
from ..psycopg import OperationalError
from ..utils import parse_int
if TYPE_CHECKING: # pragma: no cover
from psycopg import Cursor
@@ -378,10 +377,9 @@ class SlotsHandler:
except Exception:
logger.exception("Failed to create physical replication slot '%s'", name)
self._schedule_load_slots = True
elif not self._postgresql.is_primary() and self._postgresql.can_advance_slots \
and self._replication_slots[name]['type'] == 'physical':
elif self._postgresql.can_advance_slots and self._replication_slots[name]['type'] == 'physical':
value['restart_lsn'] = self._replication_slots[name]['restart_lsn']
lsn = parse_int(value.get('lsn'))
lsn = value.get('lsn')
if lsn and lsn > value['restart_lsn']: # The slot has feedback in DCS and needs to be advanced
try:
lsn = format_lsn(lsn)
@@ -477,12 +475,9 @@ class SlotsHandler:
# If the logical already exists, copy some information about it into the original structure
if name in self._replication_slots and compare_slots(value, self._replication_slots[name]):
self._copy_items(self._replication_slots[name], value)
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(value['lsn']):
advance_slots[value['database']][name] = int(value['lsn'])
except Exception as e:
logger.error('Failed to parse "%s": %r', value['lsn'], e)
if 'lsn' in value and value['confirmed_flush_lsn'] < value['lsn']: # The slot has feedback in DCS
# Skip slots that don't need to be advanced
advance_slots[value['database']][name] = value['lsn']
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)
patroni/utils.py
+1 -1
View File
@@ -820,7 +820,7 @@ def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig']
member.update({n: m.data[n] for n in optional_attributes if n in m.data})
if m.name != leader_name:
lsn = m.data.get('xlog_location')
lsn = m.lsn
if lsn is None:
member['lag'] = 'unknown'
elif cluster_lsn >= lsn:
postgres0.yml
+6 -10
View File
@@ -43,9 +43,13 @@ etcd:
# - 127.0.0.1:2223
# - 127.0.0.1:2224
# The bootstrap configuration. Works only when the cluster is not yet initialized.
# If the cluster is already initialized, all changes in the `bootstrap` section are ignored!
bootstrap:
# this section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster
# and all other cluster members will use it as a `global configuration`
# This section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster
# and all other cluster members will use it as a `global configuration`.
# WARNING! If you want to change any of the parameters that were set up
# via `bootstrap.dcs` section, please use `patronictl edit-config`!
dcs:
ttl: 30
loop_wait: 10
@@ -93,14 +97,6 @@ bootstrap:
# Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter)
# post_init: /usr/local/bin/setup_cluster.sh
# Some additional users which needs to be created after initializing new cluster
users:
admin:
password: admin%
options:
- createrole
- createdb
postgresql:
listen: 127.0.0.1:5432
connect_address: 127.0.0.1:5432
postgres1.yml
+6 -10
View File
@@ -43,9 +43,13 @@ etcd:
# - 127.0.0.1:2222
# - 127.0.0.1:2224
# The bootstrap configuration. Works only when the cluster is not yet initialized.
# If the cluster is already initialized, all changes in the `bootstrap` section are ignored!
bootstrap:
# this section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster
# and all other cluster members will use it as a `global configuration`
# This section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster
# and all other cluster members will use it as a `global configuration`.
# WARNING! If you want to change any of the parameters that were set up
# via `bootstrap.dcs` section, please use `patronictl edit-config`!
dcs:
ttl: 30
loop_wait: 10
@@ -87,14 +91,6 @@ bootstrap:
# Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter)
# post_init: /usr/local/bin/setup_cluster.sh
# Some additional users which needs to be created after initializing new cluster
users:
admin:
password: admin%
options:
- createrole
- createdb
postgresql:
listen: 127.0.0.1:5433
connect_address: 127.0.0.1:5433
postgres2.yml
+6 -10
View File
@@ -43,9 +43,13 @@ etcd:
# - 127.0.0.1:2222
# - 127.0.0.1:2223
# The bootstrap configuration. Works only when the cluster is not yet initialized.
# If the cluster is already initialized, all changes in the `bootstrap` section are ignored!
bootstrap:
# this section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster
# and all other cluster members will use it as a `global configuration`
# This section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster
# and all other cluster members will use it as a `global configuration`.
# WARNING! If you want to change any of the parameters that were set up
# via `bootstrap.dcs` section, please use `patronictl edit-config`!
dcs:
ttl: 30
loop_wait: 10
@@ -84,14 +88,6 @@ bootstrap:
- encoding: UTF8
- data-checksums
# Some additional users which needs to be created after initializing new cluster
users:
admin:
password: admin%
options:
- createrole
- createdb
postgresql:
listen: 127.0.0.1:5434
connect_address: 127.0.0.1:5434
setup.py
+25 -22
View File
@@ -26,7 +26,6 @@ KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
EXTRAS_REQUIRE = {'aws': ['boto3'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'],
'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'],
'kubernetes': [], 'raft': ['pysyncobj', 'cryptography']}
COVERAGE_XML = True
# Add here all kinds of additional classifiers as defined under
# https://pypi.python.org/pypi?%3Aaction=list_classifiers
@@ -120,14 +119,21 @@ def read(fname):
return fd.read()
def setup_package(version):
def get_versions():
old_modules = sys.modules.copy()
try:
from patroni import MIN_PSYCOPG2, MIN_PSYCOPG3
from patroni.version import __version__
return __version__, MIN_PSYCOPG2, MIN_PSYCOPG3
finally:
sys.modules.clear()
sys.modules.update(old_modules)
def main():
logging.basicConfig(format='%(message)s', level=os.getenv('LOGLEVEL', logging.WARNING))
# Assemble additional setup commands
cmdclass = {'test': PyTest, 'flake8': Flake8}
install_requires = []
for r in read('requirements.txt').split('\n'):
r = r.strip()
if r == '':
@@ -139,15 +145,22 @@ def setup_package(version):
deps[i] = r
EXTRAS_REQUIRE[e] = deps
extra = True
break
if extra:
break
if not extra:
install_requires.append(r)
# Just for convenience, if someone wants to install dependencies for all extras
EXTRAS_REQUIRE['all'] = list({e for extras in EXTRAS_REQUIRE.values() for e in extras})
patroni_version, min_psycopg2, min_psycopg3 = get_versions()
# Make it possible to specify psycopg dependency as extra
for name, version in {'psycopg[binary]': min_psycopg3, 'psycopg2': min_psycopg2, 'psycopg2-binary': None}.items():
EXTRAS_REQUIRE[name] = [name + ('>=' + '.'.join(map(str, version)) if version else '')]
EXTRAS_REQUIRE['psycopg3'] = EXTRAS_REQUIRE.pop('psycopg[binary]')
setup(
name=NAME,
version=version,
version=patroni_version,
url=URL,
author=AUTHOR,
author_email=AUTHOR_EMAIL,
@@ -163,20 +176,10 @@ def setup_package(version):
]},
install_requires=install_requires,
extras_require=EXTRAS_REQUIRE,
cmdclass=cmdclass,
cmdclass={'test': PyTest, 'flake8': Flake8},
entry_points={'console_scripts': CONSOLE_SCRIPTS},
)
if __name__ == '__main__':
old_modules = sys.modules.copy()
try:
from patroni import check_psycopg
from patroni.version import __version__
finally:
sys.modules.clear()
sys.modules.update(old_modules)
check_psycopg()
setup_package(__version__)
main()
tests/__init__.py
+1 -1
View File
@@ -241,7 +241,7 @@ class PostgresInit(unittest.TestCase):
'replication': {'username': '', 'password': 'rep-pass'},
'rewind': {'username': 'rewind', 'password': 'test'}},
'remove_data_directory_on_rewind_failure': True,
'use_pg_rewind': True, 'pg_ctl_timeout': 'bla',
'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True,
'parameters': self._PARAMETERS,
'recovery_conf': {'foo': 'bar'},
'pg_hba': ['host all all 0.0.0.0/0 md5'],
tests/test_bootstrap.py
+1 -1
View File
@@ -256,7 +256,7 @@ class TestBootstrap(BaseTestPostgresql):
mock_cancellable_subprocess_call.assert_called()
args, kwargs = mock_cancellable_subprocess_call.call_args
self.assertTrue('PGPASSFILE' in kwargs['env'])
self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=127.0.0.2 port=5432'])
self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=/tmp port=5432'])
mock_cancellable_subprocess_call.reset_mock()
self.p.connection_pool._conn_kwargs.pop('host')
tests/test_citus.py
+1 -1
View File
@@ -13,7 +13,6 @@ class TestCitus(BaseTestPostgresql):
def setUp(self):
super(TestCitus, self).setUp()
self.c = self.p.citus_handler
self.p.connection_pool.conn_kwargs = {'host': 'localhost', 'dbname': 'postgres'}
self.cluster = get_cluster_initialized_with_leader()
self.cluster.workers[1] = self.cluster
@@ -139,6 +138,7 @@ class TestCitus(BaseTestPostgresql):
self.assertEqual(parameters['max_prepared_transactions'], 202)
self.assertEqual(parameters['shared_preload_libraries'], 'citus,foo,bar')
self.assertEqual(parameters['wal_level'], 'logical')
self.assertEqual(parameters['citus.local_hostname'], '/tmp')
def test_bootstrap(self):
self.c._config = None
tests/test_etcd3.py
+11 -4
View File
@@ -6,8 +6,8 @@ import urllib3
from mock import Mock, PropertyMock, patch
from patroni.dcs.etcd import DnsCachingResolver
from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \
Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \
Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode
Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \
Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode
from threading import Thread
from . import SleepException, MockResponse
@@ -161,9 +161,16 @@ class TestPatroniEtcd3Client(BaseTestEtcd3):
mock_urlopen.return_value.content = '{"code":16,"error":"etcdserver: invalid auth token"}'
self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
with patch.object(PatroniEtcd3Client, 'authenticate', Mock(return_value=True)):
self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
retry = self.etcd3._retry.copy()
with patch('time.time', Mock(side_effect=[0, 10, 20, 30, 40])):
self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry)
self.client.username = None
self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
retry = self.etcd3._retry.copy()
self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry)
mock_urlopen.return_value.content = '{"code":3,"error":"etcdserver: revision of auth store is old"}'
self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED
self.assertRaises(AuthOldRevision, retry, self.client.deleteprefix, 'foo', retry=retry)
def test__handle_server_response(self):
response = MockResponse()
tests/test_ha.py
+2
View File
@@ -168,6 +168,7 @@ def run_async(self, func, args=()):
@patch.object(Postgresql, 'is_primary', Mock(return_value=True))
@patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1)))
@patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10))
@patch.object(Postgresql, 'slots', Mock(return_value={'l': 100}))
@patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
@patch.object(Postgresql, 'controldata', Mock(return_value={
'Database system identifier': SYSID,
@@ -1591,6 +1592,7 @@ class TestHa(PostgresInit):
@patch('patroni.psycopg.connect', psycopg_connect)
def test_permanent_logical_slots_after_promote(self):
self.p._major_version = 110000
config = ClusterConfig(1, {'slots': {'l': {'database': 'postgres', 'plugin': 'test_decoding'}}}, 1)
self.p.name = 'other'
self.ha.cluster = get_cluster_initialized_without_leader(cluster_config=config)
tests/test_patroni.py
+11 -4
View File
@@ -15,8 +15,7 @@ from patroni.dcs.etcd import AbstractEtcdClientWithFailover
from patroni.exceptions import DCSError
from patroni.postgresql import Postgresql
from patroni.postgresql.config import ConfigHandler
from patroni import check_psycopg
from patroni.__main__ import Patroni, main as _main
from patroni.__main__ import check_psycopg, Patroni, main as _main
from threading import Thread
from . import psycopg_connect, SleepException
@@ -25,10 +24,16 @@ from .test_postgresql import MockPostmaster
def mock_import(*args, **kwargs):
if args[0] == 'psycopg':
ret = Mock()
ret.__version__ = '2.5.3.dev1 a b c' if args[0] == 'psycopg2' else '3.1.0'
return ret
def mock_import2(*args, **kwargs):
if args[0] == 'psycopg2':
raise ImportError
ret = Mock()
ret.__version__ = '2.5.3.dev1 a b c'
ret.__version__ = '0.1.2'
return ret
@@ -205,6 +210,8 @@ class TestPatroni(unittest.TestCase):
with patch('builtins.__import__', Mock(side_effect=ImportError)):
self.assertRaises(SystemExit, check_psycopg)
with patch('builtins.__import__', mock_import):
self.assertIsNone(check_psycopg())
with patch('builtins.__import__', mock_import2):
self.assertRaises(SystemExit, check_psycopg)
def test_ensure_unique_name(self):
tests/test_slots.py
+1
View File
@@ -128,6 +128,7 @@ class TestSlotsHandler(BaseTestPostgresql):
self.cluster.slots['ls'] = 'a'
self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
self.cluster.config.data['slots']['ls']['database'] = 'b'
self.cluster.slots['ls'] = '500'
with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
tests/test_validator.py
+18
View File
@@ -134,6 +134,23 @@ def connect_side_effect(host_port):
raise socket.gaierror()
def mock_getaddrinfo(host, port, *args):
if port is None or port == "":
port = 0
port = int(port)
if port not in range(0, 65536):
raise socket.gaierror()
if host == "127.0.0.1" or host == "" or host is None:
return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('127.0.0.1', port))]
elif host == "127.0.0.2":
return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('127.0.0.2', port))]
elif host == "::1":
return [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('::1', port, 0, 0))]
else:
raise socket.gaierror()
def parse_output(output):
result = []
for s in output.split("\n"):
@@ -145,6 +162,7 @@ def parse_output(output):
@patch('socket.socket.connect_ex', Mock(side_effect=connect_side_effect))
@patch('socket.getaddrinfo', Mock(side_effect=mock_getaddrinfo))
@patch('os.path.exists', Mock(side_effect=exists_side_effect))
@patch('os.path.isdir', Mock(side_effect=isdir_side_effect))
@patch('os.path.isfile', Mock(side_effect=isfile_side_effect))
tests/test_zookeeper.py
+7 -15
View File
@@ -1,13 +1,13 @@
import select
import unittest
from kazoo.client import KazooClient, KazooState
from kazoo.client import KazooClient
from kazoo.exceptions import NoNodeError, NodeExistsError
from kazoo.handlers.threading import SequentialThreadingHandler
from kazoo.protocol.states import KeeperState, ZnodeStat
from kazoo.protocol.states import KeeperState, WatchedEvent, ZnodeStat
from kazoo.retry import RetryFailedError
from mock import Mock, PropertyMock, patch
from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient, \
from patroni.dcs.zookeeper import Cluster, PatroniKazooClient, \
PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
@@ -152,9 +152,6 @@ class TestZooKeeper(unittest.TestCase):
'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10,
'set_acls': {'CN=principal2': ['ALL']}})
def test_session_listener(self):
self.zk.session_listener(KazooState.SUSPENDED)
def test_reload_config(self):
self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10})
self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 5})
@@ -176,15 +173,6 @@ class TestZooKeeper(unittest.TestCase):
self.zk._cluster_loader(self.zk.client_path(''))
def test_get_cluster(self):
cluster = self.zk.get_cluster(True)
self.assertIsInstance(cluster.leader, Leader)
self.zk.status_watcher(None)
self.zk.get_cluster()
self.zk.touch_member({'foo': 'foo'})
self.zk._name = 'bar'
self.zk.status_watcher(None)
with patch.object(ZooKeeper, 'get_node', Mock(side_effect=Exception)):
self.zk.get_cluster()
cluster = self.zk.get_cluster()
self.assertEqual(cluster.last_lsn, 500)
@@ -295,3 +283,7 @@ class TestZooKeeper(unittest.TestCase):
def test_set_history_value(self):
self.zk.set_history_value('{}')
def test_watcher(self):
self.zk._watcher(WatchedEvent('', '', ''))
self.assertTrue(self.zk.watch(1, 1))