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

Don't touch logical failover slots (#3245)

Browse Source 
If logical replication slot is created with failover => true option, we
get respective field set to true in `pg_replication_slots` view.

By avoiding interacting with such slots we make logical failover slots
feature fully functional in PG17.
This commit is contained in:
Alexander Kukushkin
2025-02-14 08:35:37 +01:00
committed by GitHub
parent 1a31ea6e20
commit 0d87270897
9 changed files with 48 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/dcs_failsafe_mode.feature
+3 -3
View File
@@ -42,7 +42,7 @@ Feature: dcs failsafe mode
And I receive a response postgres-1 http://127.0.0.1:8009/patroni
@dcs-failsafe
@slot-advance
@pg110000
Scenario: check leader and replica are functioning while DCS is down
Given I get all changes from physical slot dcs_slot_1 on postgres-0
Then physical slot dcs_slot_1 is in sync between postgres-0 and postgres-1 after 10 seconds
@@ -84,7 +84,7 @@ Feature: dcs failsafe mode
And replication works from postgres-1 to postgres-2 after 10 seconds
@dcs-failsafe
@slot-advance
@pg110000
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 postgres-1 and postgres-0 after 20 seconds
@@ -104,7 +104,7 @@ Feature: dcs failsafe mode
And postgres-2 role is the replica after 2 seconds
@dcs-failsafe
@slot-advance
@pg110000
Scenario: check that permanent slots are in sync between nodes while DCS is down
Given replication works from postgres-1 to postgres-0 after 10 seconds
And replication works from postgres-1 to postgres-2 after 10 seconds
features/environment.py
+10 -3
View File
@@ -1153,10 +1153,17 @@ def after_feature(context, feature):
def before_scenario(context, scenario):
if 'slot-advance' in scenario.effective_tags:
for tag in scenario.effective_tags:
if tag.startswith('pg') and 6 < len(tag) < 9:
try:
ver = int(tag[2:])
except Exception:
ver = 0
if not ver:
continue
for p in context.pctl._processes.values():
if p._conn and p._conn.server_version < 110000:
scenario.skip('pg_replication_slot_advance() is not supported on {0}'.format(p._conn.server_version))
if p._conn and p._conn.server_version < ver:
scenario.skip('not supported on {0}'.format(p._conn.server_version))
break
if 'dcs-failsafe' in scenario.effective_tags and not context.dcs_ctl._handle:
scenario.skip('it is not possible to control state of {0} from tests'.format(context.dcs_ctl.name()))
features/ignored_slots.feature
+9 -4
View File
@@ -6,11 +6,10 @@ Feature: ignored slots
When I issue a PATCH request to http://127.0.0.1:8009/config with {"ignore_slots": [{"name": "unmanaged_slot_0", "database": "postgres", "plugin": "test_decoding", "type": "logical"}, {"name": "unmanaged_slot_1", "database": "postgres", "plugin": "test_decoding"}, {"name": "unmanaged_slot_2", "database": "postgres"}, {"name": "unmanaged_slot_3"}], "postgresql": {"parameters": {"wal_level": "logical"}}}
Then I receive a response code 200
And Response on GET http://127.0.0.1:8009/config contains ignore_slots after 10 seconds
And Response on GET http://127.0.0.1:8009/patroni contains pending_restart after 10 seconds
# Make sure the wal_level has been changed.
When I shut down postgres-1
And I start postgres-1
Then postgres-1 is a leader after 10 seconds
And "members/postgres-1" key in DCS has role=primary after 10 seconds
When I run patronictl.py restart batman postgres-1 --force
Then "members/postgres-1" key in DCS has role=primary after 10 seconds
# Make sure Patroni has finished telling Postgres it should be accepting writes.
And postgres-1 role is the primary after 20 seconds
# 1. Create our test logical replication slot.
@@ -59,3 +58,9 @@ Feature: ignored slots
And postgres-1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
And postgres-1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
And postgres-1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds
@pg170000
Scenario: check that logical slots with failover are not removed by Patroni
Given I create a logical failover slot test17 on postgres-1 with the pgoutput plugin
When I run patronictl.py restart batman postgres-1 --force
Then postgres-1 has a logical replication slot named test17 with the pgoutput plugin after 2 seconds
features/nostream_node.feature
+2 -2
View File
@@ -6,7 +6,7 @@ Scenario: check nostream node is recovering from archive
Then "members/postgres-1" key in DCS has replication_state=in archive recovery after 10 seconds
And replication works from postgres-0 to postgres-1 after 30 seconds
@slot-advance
@pg110000
Scenario: check permanent logical replication slots are not copied
When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}}, "slots":{"test_logical":{"type":"logical","database":"postgres","plugin":"test_decoding"}}}
Then I receive a response code 200
@@ -17,7 +17,7 @@ Scenario: check permanent logical replication slots are not copied
And postgres-1 does not have a replication slot named test_logical
And postgres-2 does not have a replication slot named test_logical
@slot-advance
@pg110000
Scenario: check that slots are written to the /status key
Given "status" key in DCS has postgres_0 in slots
And "status" key in DCS has postgres_2 in slots
features/permanent_slots.feature
+6 -6
View File
@@ -15,13 +15,13 @@ Feature: permanent slots
And postgres-2 has a physical replication slot named postgres_3 after 10 seconds
And postgres-2 does not have a replication slot named test_physical
@slot-advance
@pg110000
Scenario: check that logical permanent slots are created
Given I run patronictl.py restart batman postgres-0 --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"}}}
Then postgres-0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
@slot-advance
@pg110000
Scenario: check that permanent slots are created on replicas
Given postgres-1 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 postgres-0 and postgres-1 after 10 seconds
@@ -30,7 +30,7 @@ Feature: permanent slots
And postgres-2 does not have a replication slot named test_logical
And postgres-3 has a physical replication slot named test_physical after 2 seconds
@slot-advance
@pg110000
Scenario: check permanent physical slots that match with member names
Given postgres-0 has a physical replication slot named postgres_3 after 2 seconds
And postgres-1 has a physical replication slot named postgres_0 after 2 seconds
@@ -43,7 +43,7 @@ Feature: permanent slots
And postgres-3 has a physical replication slot named postgres_1 after 2 seconds
And postgres-3 has a physical replication slot named postgres_2 after 2 seconds
@slot-advance
@pg110000
Scenario: check that permanent slots are advanced on replicas
Given I add the table replicate_me to postgres-0
When I get all changes from logical slot test_logical on postgres-0
@@ -56,7 +56,7 @@ Feature: permanent slots
And Physical slot postgres_3 is in sync between postgres-2 and postgres-0 after 20 seconds
And Physical slot postgres_3 is in sync between postgres-2 and postgres-1 after 10 seconds
@slot-advance
@pg110000
Scenario: check that permanent slots and member slots are written to the /status key
Given "status" key in DCS has test_physical in slots
And "status" key in DCS has postgres_0 in slots
@@ -64,7 +64,7 @@ Feature: permanent slots
And "status" key in DCS has postgres_2 in slots
And "status" key in DCS has postgres_3 in slots
@slot-advance
@pg110000
Scenario: check that only non-permanent member slots are written to the retain_slots in /status key
Given "status" key in DCS has postgres_0 in retain_slots
And "status" key in DCS has postgres_1 in retain_slots
features/standby_cluster.feature
+1 -1
View File
@@ -16,7 +16,7 @@ Feature: standby cluster
And Response on GET http://127.0.0.1:8008/patroni contains replication_state=streaming after 10 seconds
And "members/postgres-0" key in DCS has replication_state=streaming after 10 seconds
@slot-advance
@pg110000
Scenario: check permanent logical slots are synced to the replica
Given I run patronictl.py restart batman postgres-1 --force
Then Logical slot test_logical is in sync between postgres-0 and postgres-1 after 10 seconds
features/steps/slots.py
+5 -7
View File
@@ -6,15 +6,13 @@ from behave import step, then
import patroni.psycopg as pg
@step('I create a logical replication slot {slot_name} on {pg_name:name} with the {plugin:w} plugin')
def create_logical_replication_slot(context, slot_name, pg_name, plugin):
@step('I create a logical {slot_type} slot {slot_name} on {pg_name:name} with the {plugin:w} plugin')
def create_logical_replication_slot(context, slot_type, slot_name, pg_name, plugin):
failover = ', failover=>true' if slot_type == 'failover' else ''
try:
output = context.pctl.query(pg_name, ("SELECT pg_create_logical_replication_slot('{0}', '{1}'),"
" current_database()").format(slot_name, plugin))
print(output.fetchone())
context.pctl.query(pg_name, f"SELECT pg_create_logical_replication_slot('{slot_name}', '{plugin}'{failover})")
except pg.Error as e:
print(e)
assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
assert False, "Error creating slot {0} on {1} with plugin {2}: {3}".format(slot_name, pg_name, plugin, e)
@step('{pg_name:name} has a logical replication slot named {slot_name}'
patroni/postgresql/__init__.py
+2 -1
View File
@@ -228,11 +228,12 @@ class Postgresql(object):
and self.role in ('primary', 'promoted') else "'on', '', NULL")
if self._major_version >= 90600:
filter_failover = ' WHERE NOT failover' if self._major_version >= 170000 else ''
extra = ("pg_catalog.current_setting('restore_command')" if self._major_version >= 120000 else "NULL") +\
", " + ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
"plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
" AS confirmed_flush_lsn, pg_catalog.pg_wal_lsn_diff(restart_lsn, '0/0')::bigint"
" AS restart_lsn, xmin FROM pg_catalog.pg_get_replication_slots()) AS s)"
f" AS restart_lsn, xmin FROM pg_catalog.pg_get_replication_slots(){filter_failover}) AS s)"
if self._should_query_slots and self.can_advance_slots else "NULL") + extra
extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
" slot_name, conninfo, status, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
patroni/postgresql/slots.py
+8 -3
View File
@@ -273,10 +273,13 @@ class SlotsHandler:
pg_wal_lsn_diff = f"pg_catalog.pg_{self._postgresql.wal_name}_{self._postgresql.lsn_name}_diff"
extra = f", catalog_xmin, {pg_wal_lsn_diff}(confirmed_flush_lsn, '0/0')::bigint" \
if self._postgresql.major_version >= 100000 else ""
skip_temp_slots = ' WHERE NOT temporary' if self._postgresql.major_version >= 100000 else ''
filter_columns = tuple(fltr for fltr, major in (('temporary', 100000), ('failover', 170000))
if self._postgresql.major_version >= major)
where_filter = ' AND '.join(map(lambda col: f'NOT {col}', filter_columns))
where_condition = f' WHERE {where_filter}' if where_filter else ''
for r in self._query("SELECT slot_name, slot_type, xmin, "
f"{pg_wal_lsn_diff}(restart_lsn, '0/0')::bigint, plugin, database, datoid{extra}"
f" FROM pg_catalog.pg_replication_slots{skip_temp_slots}"):
f" FROM pg_catalog.pg_replication_slots{where_condition}"):
value = {'type': r[1]}
if r[1] == 'logical':
value.update(plugin=r[4], database=r[5], datoid=r[6])
@@ -694,11 +697,13 @@ class SlotsHandler:
copy_slots: Dict[str, Dict[str, Any]] = {}
with self._get_leader_connection_cursor(leader) as cur:
try:
filter_failover = ' NOT failover AND' if self._postgresql.major_version >= 170000 else ''
cur.execute("SELECT slot_name, slot_type, datname, plugin, catalog_xmin, "
"pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint, "
"pg_catalog.pg_read_binary_file('pg_replslot/' || slot_name || '/state')"
" FROM pg_catalog.pg_get_replication_slots() JOIN pg_catalog.pg_database ON datoid = oid"
" WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)", (create_slots,))
f" WHERE{filter_failover} NOT pg_catalog.pg_is_in_recovery()"
" AND slot_name = ANY(%s)", (create_slots,))
for r in cur:
if r[0] in slots: # slot_name is defined in the global configuration