diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 805ce4c2..f496ec25 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -38,6 +38,32 @@ Dynamic configuration is stored in the DCS (Distributed Configuration Store) and - **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. - **plugin**: the plugin name for the logical slot. +- **ignore_slots**: list of sets of replication slot properties for which Patroni should ignore matching slots. This configuration/feature/etc. is useful when some replication slots are managed outside of Patroni. Any subset of matching properties will cause a slot to be ignored. + - **name**: the name of the replication slot. + - **type**: slot type. Can be ``physical`` or ``logical``. If the slot is logical, you may additionally define ``database`` and/or ``plugin``. + - **database**: the database name (when matching a ``logical`` slot). + - **plugin**: the logical decoding plugin (when matching a ``logical`` slot). + +Note: **slots** is a hashmap while **ignore_slots** is an array. For example: + +.. code:: YAML + + slots: + permanent_logical_slot_name: + type: logical + database: my_db + plugin: test_decoding + permanent_physical_slot_name: + type: physical + ... + ignore_slots: + - name: ignored_logical_slot_name + type: logical + database: my_db + plugin: test_decoding + - name: ignored_physical_slot_name + type: physical + ... Global/Universal ---------------- diff --git a/features/ignored_slots.feature b/features/ignored_slots.feature new file mode 100644 index 00000000..cb6dbf79 --- /dev/null +++ b/features/ignored_slots.feature @@ -0,0 +1,61 @@ +Feature: ignored slots + Scenario: check ignored slots aren't removed on failover/switchover + Given I start postgres1 + Then postgres1 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:8009/config with {"loop_wait": 2, "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 + # Make sure the wal_level has been changed. + When I shut down postgres1 + And I start postgres1 + Then postgres1 is a leader after 10 seconds + And "members/postgres1" key in DCS has role=master after 3 seconds + # Make sure Patroni has finished telling Postgres it should be accepting writes. + And postgres1 role is the primary after 20 seconds + # 1. Create our test logical replication slot. + # Test that ny subset of attributes in the ignore slots matcher is enough to match a slot + # by using 3 different slots. + When I create a logical replication slot unmanaged_slot_0 on postgres1 with the test_decoding plugin + And I create a logical replication slot unmanaged_slot_1 on postgres1 with the test_decoding plugin + And I create a logical replication slot unmanaged_slot_2 on postgres1 with the test_decoding plugin + And I create a logical replication slot unmanaged_slot_3 on postgres1 with the test_decoding plugin + And I create a logical replication slot dummy_slot on postgres1 with the test_decoding plugin + # It seems like it'd be obvious that these slots exist since we just created them, + # but Patroni can actually end up dropping them almost immediately, so it's helpful + # to verify they exist before we begin testing whether they persist through failover + # cycles. + Then postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin + + When I start postgres0 + Then "members/postgres0" key in DCS has role=replica after 3 seconds + And postgres0 role is the secondary after 20 seconds + # Verify that the replica has advanced beyond the point in the WAL + # where we created the replication slot so that on the next failover + # cycle we don't accidentally rewind to before the slot creation. + And replication works from postgres1 to postgres0 after 20 seconds + When I shut down postgres1 + Then "members/postgres0" key in DCS has role=master after 3 seconds + + # 2. After a failover the server (now a replica) still has the slot. + When I start postgres1 + Then postgres1 role is the secondary after 20 seconds + And "members/postgres1" key in DCS has role=replica after 3 seconds + # give Patroni time to sync replication slots + And I sleep for 2 seconds + And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin + And postgres1 does not have a logical replication slot named dummy_slot + + # 3. After a failover the server (now a master) still has the slot. + When I shut down postgres0 + Then "members/postgres1" key in DCS has role=master after 3 seconds + And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin + And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin diff --git a/features/steps/cascading_replication.py b/features/steps/cascading_replication.py index 299e8622..d979a311 100644 --- a/features/steps/cascading_replication.py +++ b/features/steps/cascading_replication.py @@ -12,8 +12,10 @@ def start_patroni_with_a_name_value_tag(context, name, tag_name, tag_value): @then('There is a {label} with "{content}" in {name:w} data directory') def check_label(context, label, content, name): label = context.pctl.read_label(name, label) + if label is None: + label = "" label = label.replace('\n', '\\n') - assert content in label, "{0} doesn't contain {1}".format(label, content) + assert content in label, "\"{0}\" doesn't contain {1}".format(label, content) @step('I create label with "{content:w}" in {name:w} data directory') @@ -25,15 +27,17 @@ def write_label(context, content, name): def check_member(context, name, key, value, time_limit): time_limit *= context.timeout_multiplier max_time = time.time() + int(time_limit) + dcs_value = None while time.time() < max_time: try: response = json.loads(context.dcs_ctl.query(name)) - if response.get(key) == value: + dcs_value = response.get(key) + if dcs_value == value: return except Exception: pass time.sleep(1) - assert False, "{0} does not have {1}={2} in dcs after {3} seconds".format(name, key, value, time_limit) + assert False, "{0} does not have {1}={2} (found {3}) in dcs after {4} seconds".format(name, key, value, dcs_value, time_limit) @step('there is a non empty {key:w} key in DCS after {time_limit:d} seconds') diff --git a/features/steps/slots.py b/features/steps/slots.py new file mode 100644 index 00000000..c6ce624f --- /dev/null +++ b/features/steps/slots.py @@ -0,0 +1,30 @@ +from behave import step, then +import psycopg2 as pg + + +@step('I create a logical replication slot {slot_name} on {pg_name:w} with the {plugin:w} plugin') +def create_logical_replication_slot(context, slot_name, pg_name, plugin): + try: + output = context.pctl.query(pg_name, "SELECT pg_create_logical_replication_slot('{0}', '{1}'), current_database()".format(slot_name, plugin)) + print(output.fetchone()) + except pg.Error as e: + print(e) + assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin) + +@then('{pg_name:w} has a logical replication slot named {slot_name} with the {plugin:w} plugin') +def has_logical_replication_slot(context, pg_name, slot_name, plugin): + try: + row = context.pctl.query(pg_name, "SELECT slot_type, plugin FROM pg_replication_slots WHERE slot_name = '{0}'".format(slot_name)).fetchone() + assert row, "Couldn't find replication slot named {0}".format(slot_name) + assert row[0] == "logical", "Found replication slot named {0} but wasn't a logical slot".format(slot_name) + assert row[1] == plugin, "Found replication slot named {0} but was using plugin {1} rather than {2}".format(slot_name, row[1], plugin) + except pg.Error as e: + assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin) + +@then('{pg_name:w} does not have a logical replication slot named {slot_name}') +def has_logical_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() + assert not row, "Found unexpected replication slot named {0}".format(slot_name) + except pg.Error as e: + assert False, "Error looking for slot {0} on {1}".format(slot_name, pg_name) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index a118f8e6..e5503f3e 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -341,6 +341,10 @@ class ClusterConfig(namedtuple('ClusterConfig', 'index,data,modify_index')): self.data.get('permanent_slots') or self.data.get('slots') ) or {} + @property + def ignore_slots_matchers(self): + return isinstance(self.data, dict) and self.data.get('ignore_slots') or [] + @property def max_timelines_history(self): return self.data.get('max_timelines_history', 0) diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index fac8eced..4ba425ce 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -33,6 +33,14 @@ class SlotsHandler(object): self._replication_slots = replication_slots self._schedule_load_slots = False + def ignore_replication_slot(self, cluster, name): + slot = self._replication_slots[name] + for matcher in cluster.config.ignore_slots_matchers: + if ((matcher.get("name") is None or matcher["name"] == name) + and all(not matcher.get(a) or matcher[a] == slot.get(a) for a in ('database', 'plugin', 'type'))): + return True + return False + def drop_replication_slot(self, name): cursor = self._query(('SELECT pg_catalog.pg_drop_replication_slot(%s) WHERE EXISTS (SELECT 1 ' + 'FROM pg_catalog.pg_replication_slots WHERE slot_name = %s AND NOT active)'), name, name) @@ -48,7 +56,7 @@ class SlotsHandler(object): # drop old replication slots which are not presented in desired slots for name in set(self._replication_slots) - set(slots): - if not self.drop_replication_slot(name): + if not self.ignore_replication_slot(cluster, name) and not self.drop_replication_slot(name): logger.error("Failed to drop replication slot '%s'", name) self._schedule_load_slots = True