Don't copy logical slot if there is mismatch with the config (#2274)

A couple of times we have seen in the wild that the database for the permanent logical slots was changed in the Patroni config.

It resulted in the below situation.
On the primary:
1. The slot must be dropped before creating it in a different DB.
2. Patroni fails to drop it because the slot is in use.

Replica:
1. Patroni notice that the slot exists in the wrong DB and successfully dropping it.
2. Patroni copying the existing slot from the primary by its name with Postgres restart.

And the loop repeats while the "wrong" slot exists on the primary.

Basically, replicas are continuously restarting, which badly affects availability.

In order to solve the problem, we will perform additional checks while copying replication slot files from the primary and discard them if `slot_type`, `database`, or `plugin` don't match our expectations.

patroni/ha.py

@@ -1490,7 +1490,7 @@ class Ha(object):
                     if create_slots and self.cluster.leader:
                         err = self._async_executor.try_run_async('copy_logical_slots',
                                                                  self.state_handler.slots_handler.copy_logical_slots,
-                                                                 args=(self.cluster.leader, create_slots))
+                                                                 args=(self.cluster, create_slots))
                         if not err:
                             ret = 'Copying logical slots {0} from the primary'.format(create_slots)
             return ret

patroni/postgresql/slots.py

@@ -259,21 +259,33 @@ class SlotsHandler(object):
                 if value:
                     logger.info('Logical slot %s is safe to be used after a failover', name)
 
-    def copy_logical_slots(self, leader, slots):
+    def copy_logical_slots(self, cluster, create_slots):
+        leader = cluster.leader
+        slots = cluster.get_replication_slots(self._postgresql.name, 'replica', False, self._postgresql.major_version)
         with self._get_leader_connection_cursor(leader) as cur:
             try:
-                cur.execute("SELECT slot_name, catalog_xmin, "
+                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() WHERE NOT pg_catalog.pg_is_in_recovery()"
-                            " AND slot_name = ANY(%s)", (slots,))
-                slots = {r[0]: {'catalog_xmin': r[1], 'confirmed_flush_lsn': r[2], 'data': r[3]} for r in cur}
+                            " 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,))
+
+                create_slots = {}
+                for r in cur:
+                    if r[0] in slots:  # slot_name is defined in the global configuration
+                        slot = {'type': r[1], 'database': r[2], 'plugin': r[3],
+                                'catalog_xmin': r[4], 'confirmed_flush_lsn': r[5], 'data': r[6]}
+                        if compare_slots(slot, slots[r[0]]):
+                            create_slots[r[0]] = slot
+                        else:
+                            logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: ' +
+                                           'configuration=%s, slot on the primary=%s', r[0], slots[r[0]], slot)
             except Exception as e:
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
-        if isinstance(slots, dict) and self._postgresql.stop():
+        if isinstance(create_slots, dict) and create_slots and self._postgresql.stop():
             pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
-            for name, value in slots.items():
+            for name, value in create_slots.items():
                 slot_dir = os.path.join(pg_replslot_dir, name)
                 slot_tmp_dir = slot_dir + '.tmp'
                 if os.path.exists(slot_tmp_dir):

tests/__init__.py

@@ -93,8 +93,8 @@ class MockCursor(object):
             raise RetryFailedError('retry')
         elif sql.startswith('SELECT catalog_xmin'):
             self.results = [(100, 501)]
-        elif sql.startswith('SELECT slot_name, catalog_xmin'):
-            self.results = [('ls', 100, 500, b'123456')]
+        elif sql.startswith('SELECT slot_name, slot_type, datname, plugin, catalog_xmin'):
+            self.results = [('ls', 'logical', 'a', 'b', 100, 500, b'123456')]
         elif sql.startswith('SELECT slot_name'):
             self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)]
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):

tests/test_ha.py

@@ -1211,9 +1211,12 @@ class TestHa(PostgresInit):
     @patch('os.rename', Mock())
     @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False))
     @patch.object(builtins, 'open', mock_open())
-    @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['foo']))
+    @patch.object(ConfigHandler, 'check_recovery_conf', Mock(return_value=(False, False)))
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000))
+    @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['ls']))
     def test_follow_copy(self):
         self.ha.cluster.is_unlocked = false
+        self.ha.cluster.config.data['slots'] = {'ls': {'database': 'a', 'plugin': 'b'}}
         self.p.is_leader = false
         self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots'))
 

tests/test_slots.py

@@ -27,6 +27,9 @@ class TestSlotsHandler(BaseTestPostgresql):
         super(TestSlotsHandler, self).setUp()
         self.s = self.p.slots_handler
         self.p.start()
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        self.cluster = Cluster(True, config, self.leader, 0,
+                               [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
 
     def test_sync_replication_slots(self):
         config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
@@ -81,41 +84,38 @@ class TestSlotsHandler(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        self.cluster.slots['ls'] = 12346
+        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         self.s._schedule_load_slots = False
         with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
-        cluster.slots['ls'] = 'a'
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
+        self.cluster.slots['ls'] = 'a'
+        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
 
-    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError))
     def test_copy_logical_slots(self):
-        self.s.copy_logical_slots(self.leader, ['foo'])
+        self.cluster.config.data['slots']['ls']['database'] = 'b'
+        self.s.copy_logical_slots(self.cluster, ['ls'])
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)):
+            self.s.copy_logical_slots(self.cluster, ['foo'])
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test_check_logical_slots_readiness(self):
-        self.s.copy_logical_slots(self.leader, ['ls'])
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        self.s.copy_logical_slots(self.cluster, ['ls'])
+        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
-            self.s.check_logical_slots_readiness(cluster, False, None)
+            self.s.check_logical_slots_readiness(self.cluster, False, None)
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test_on_promote(self):
-        self.s.copy_logical_slots(self.leader, ['ls'])
+        self.s.copy_logical_slots(self.cluster, ['ls'])
         self.s.on_promote()
 
     @unittest.skipIf(os.name == 'nt', "Windows not supported")