Feature: ability to ignore externally managed replication slots (#1742)

There are sometimes good reasons to manage replication slots externally
to Patroni. For example, a consumer may wish to manage its own slots (so
that it can more easily track when a failover has a occurred and whether
it is ahead of or behind the WAL position on the new primary).
Additionally tooling like pglogical actually replicates slots to all
replicas so that the current position can be maintained on failover
targets (this also aids consumers by supplying primitives so that they
can verify data hasn't been lost or a split brain occurred relative to
the physical cluster).

To support these use cases this new feature allows configuring Patroni
to entirely ignore sets of slots specified by any subset of name,
database, slot type, and plugin.

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
 ----------------

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

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')

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)

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)

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