diff --git a/features/dcs_failsafe_mode.feature b/features/dcs_failsafe_mode.feature index ca36e899..45218a7f 100644 --- a/features/dcs_failsafe_mode.feature +++ b/features/dcs_failsafe_mode.feature @@ -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 diff --git a/features/environment.py b/features/environment.py index 9f9adfd2..0ddcf306 100644 --- a/features/environment.py +++ b/features/environment.py @@ -1153,11 +1153,18 @@ def after_feature(context, feature): def before_scenario(context, scenario): - if 'slot-advance' in scenario.effective_tags: - 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)) - break + 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 < 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())) if 'reject-duplicate-name' in scenario.effective_tags and context.dcs_ctl.name() == 'raft': diff --git a/features/ignored_slots.feature b/features/ignored_slots.feature index 422c650b..12628d3d 100644 --- a/features/ignored_slots.feature +++ b/features/ignored_slots.feature @@ -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 diff --git a/features/nostream_node.feature b/features/nostream_node.feature index da9d4163..04502b3e 100644 --- a/features/nostream_node.feature +++ b/features/nostream_node.feature @@ -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 diff --git a/features/permanent_slots.feature b/features/permanent_slots.feature index 501c48e2..dbea3c00 100644 --- a/features/permanent_slots.feature +++ b/features/permanent_slots.feature @@ -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 diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature index c4c2aa6b..2a76471e 100644 --- a/features/standby_cluster.feature +++ b/features/standby_cluster.feature @@ -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 diff --git a/features/steps/slots.py b/features/steps/slots.py index befd1007..97bcdabe 100644 --- a/features/steps/slots.py +++ b/features/steps/slots.py @@ -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}' diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 7a0e7a2a..a720af51 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -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) diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 8af6bf07..9420c68e 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -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