Repository: zalando/patroni
Pull request #2060 (2021-09-17 06:41:49)

Ensure pg_replication_slot_advance() doesn't timeout
The bigger gap between the slot flush LSN and the LSN we want to advance to becomes more time it takes for the call to finish.
Once started failing the "lag" will grow more or less infinitely, that have the following negative side-effects:
1. Size of pg_wal on the replica will grow
2. Size of pg_wal on the primary will grow
3. Since the hot_standby_feedback is forcefully enabled, the primary will stop cleaning up dead tuples
I.e., we are not only in danger of running out of disk space, but also increasing chances of transaction wraparound to happen.

In order to mitigate it, we want to set the `statement_timeout` to 0 before calling `pg_replication_slot_advance()`.

Since the call is happening from the main HA loop and could take more than `loop_wait`, the next heartbeat run could be delayed.
There is also a possibility that the call could take longer than `ttl` and the member key/session in DCS for a given replica expires, but, the slot LSN in DCS is updated by the primary every `loop_wait` seconds. Hence, we don't expect that the slot_advance() call will take significantly longer than the `loop_wait` and therefore chances of the member key/session to expire are very low.


Patch:
diff --git a/patroni/ha.py b/patroni/ha.py
index 296dae73..209a5218 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -826,6 +826,8 @@ class Ha(object):
             'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async_req=True),
         }[mode]
 
+        logger.info('Demoting self (%s)', mode)
+
         self._rewind.trigger_check_diverged_lsn()
         self.state_handler.stop(mode_control['stop'], checkpoint=mode_control['checkpoint'],
                                 on_safepoint=self.watchdog.disable if self.watchdog.is_running else None,
@@ -997,13 +999,13 @@ class Ha(object):
                     # in case of standby cluster we don't really need to
                     # enforce anything, since the leader is not a master.
                     # So just remind the role.
-                    msg = 'no action. I am ({0}) the standby leader with the lock'.format(self.state_handler.name) \
+                    msg = 'no action. I am ({0}), the standby leader with the lock'.format(self.state_handler.name) \
                           if self.state_handler.role == 'standby_leader' else \
                           'promoted self to a standby leader because i had the session lock'
                     return self.enforce_follow_remote_master(msg)
                 else:
                     return self.enforce_master_role(
-                        'no action. I am ({0}) the leader with the lock'.format(self.state_handler.name),
+                        'no action. I am ({0}), the leader with the lock'.format(self.state_handler.name),
                         'promoted self to leader because I had the session lock'
                     )
             else:
@@ -1021,10 +1023,10 @@ class Ha(object):
         lock_owner = self.cluster.leader and self.cluster.leader.name
         if self.is_standby_cluster():
             return self.follow('cannot be a real primary in a standby cluster',
-                               'no action. I am a secondary ({0}) and following a standby leader ({1})'.format(
+                               'no action. I am ({0}), a secondary, and following a standby leader ({1})'.format(
                                     self.state_handler.name, lock_owner), refresh=False)
         return self.follow('demoting self because I do not have the lock and I was a leader',
-                           'no action. I am a secondary ({0}) and following a leader ({1})'.format(
+                           'no action. I am ({0}), a secondary, and following a leader ({1})'.format(
                                 self.state_handler.name, lock_owner), refresh=False)
 
     def evaluate_scheduled_restart(self):
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index f1ce1ecb..6cfd42c3 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -861,20 +861,22 @@ class Postgresql(object):
         if change_role:
             self.__cb_pending = ACTION_NOOP
 
+        ret = True
         if self.is_running():
             if do_reload:
                 self.config.write_postgresql_conf()
-                if self.reload(block_callbacks=change_role) and change_role:
+                ret = self.reload(block_callbacks=change_role)
+                if ret and change_role:
                     self.set_role(role)
             else:
-                self.restart(block_callbacks=change_role, role=role)
+                ret = self.restart(block_callbacks=change_role, role=role)
         else:
-            self.start(timeout=timeout, block_callbacks=change_role, role=role)
+            ret = self.start(timeout=timeout, block_callbacks=change_role, role=role) or None
 
         if change_role:
             # TODO: postpone this until start completes, or maybe do even earlier
             self.call_nowait(ACTION_ON_ROLE_CHANGE)
-        return True
+        return ret
 
     def _wait_promote(self, wait_seconds):
         for _ in polling_loop(wait_seconds):
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index 43b7e30c..015108ac 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -82,8 +82,9 @@ class SlotsHandler(object):
             replication_slots = {}
             extra = ", catalog_xmin, pg_catalog.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 ''
             cursor = self._query('SELECT slot_name, slot_type, plugin, database, datoid'
-                                 '{0} FROM pg_catalog.pg_replication_slots'.format(extra))
+                                 '{0} FROM pg_catalog.pg_replication_slots{1}'.format(extra, skip_temp_slots))
             for r in cursor:
                 value = {'type': r[1]}
                 if r[1] == 'logical':
@@ -142,9 +143,9 @@ class SlotsHandler(object):
                 self._schedule_load_slots = True
 
     @contextmanager
-    def _get_local_connection_cursor(self, database):
+    def _get_local_connection_cursor(self, **kwargs):
         conn_kwargs = self._postgresql.config.local_connect_kwargs
-        conn_kwargs['database'] = database
+        conn_kwargs.update(kwargs)
         with get_connection_cursor(**conn_kwargs) as cur:
             yield cur
 
@@ -161,7 +162,7 @@ class SlotsHandler(object):
 
         # Create new logical slots
         for database, values in logical_slots.items():
-            with self._get_local_connection_cursor(database) as cur:
+            with self._get_local_connection_cursor(database=database) as cur:
                 for name, value in values.items():
                     try:
                         cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" +
@@ -193,7 +194,7 @@ class SlotsHandler(object):
 
         # Advance logical slots
         for database, values in advance_slots.items():
-            with self._get_local_connection_cursor(database) as cur:
+            with self._get_local_connection_cursor(database=database, options='-c statement_timeout=0') as cur:
                 for name, value in values.items():
                     try:
                         cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)",


Test patch:
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 6d0b2d64..3228be69 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -349,7 +349,7 @@ class TestHa(PostgresInit):
         self.ha.has_lock = true
         self.p.is_leader = false
         self.p.set_role('master')
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
     def test_demote_after_failing_to_obtain_lock(self):
         self.ha.acquire_lock = false
@@ -389,7 +389,7 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.ha.cluster.is_unlocked = false
         self.ha.has_lock = true
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
     def test_demote_because_not_having_lock(self):
         self.ha.cluster.is_unlocked = false
@@ -408,16 +408,16 @@ class TestHa(PostgresInit):
     def test_follow(self):
         self.ha.cluster.is_unlocked = false
         self.p.is_leader = false
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()')
         self.ha.patroni.replicatefrom = "foo"
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
         self.ha.cluster.config.data.update({'slots': {'l': {'database': 'a', 'plugin': 'b'}}})
         self.ha.cluster.members[1].data['tags']['replicatefrom'] = 'postgresql0'
         self.ha.patroni.nofailover = True
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()')
         del self.ha.cluster.config.data['slots']
         self.ha.cluster.config.data.update({'postgresql': {'use_slots': False}})
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), a secondary, and following a leader ()')
         del self.ha.cluster.config.data['postgresql']['use_slots']
 
     def test_follow_in_pause(self):
@@ -536,27 +536,27 @@ class TestHa(PostgresInit):
         self.ha.fetch_node_status = get_node_status()
         self.ha.has_lock = true
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
         f = Failover(0, self.p.name, '', None)
         self.ha.cluster = get_cluster_initialized_with_leader(f)
         self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
         self.ha._rewind.rewind_or_reinitialize_needed_and_possible = true
         self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
         self.ha.fetch_node_status = get_node_status(nofailover=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
         self.ha.fetch_node_status = get_node_status(watchdog_failed=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
         self.ha.fetch_node_status = get_node_status(timeline=1)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
         self.ha.fetch_node_status = get_node_status(wal_position=1)
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
         # manual failover from the previous leader to us won't happen if we hold the nofailover flag
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
         # Failover scheduled time must include timezone
         scheduled = datetime.datetime.now()
@@ -565,28 +565,28 @@ class TestHa(PostgresInit):
 
         scheduled = datetime.datetime.utcnow().replace(tzinfo=tzutc)
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
 
         scheduled = scheduled + datetime.timedelta(seconds=30)
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
 
         scheduled = scheduled + datetime.timedelta(seconds=-600)
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
 
         scheduled = None
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
 
     def test_manual_failover_from_leader_in_pause(self):
         self.ha.has_lock = true
         self.ha.is_paused = true
         scheduled = datetime.datetime.now()
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('PAUSE: no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
-        self.assertEqual('PAUSE: no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('PAUSE: no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
 
     def test_manual_failover_from_leader_in_synchronous_mode(self):
         self.p.is_leader = true
@@ -595,7 +595,7 @@ class TestHa(PostgresInit):
         self.ha.is_failover_possible = false
         self.ha.process_sync_replication = Mock()
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, None))
-        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0), the leader with the lock', self.ha.run_cycle())
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, 'a'))
         self.ha.is_failover_possible = true
         self.assertEqual('manual failover: demoting myself', self.ha.run_cycle())
@@ -756,7 +756,7 @@ class TestHa(PostgresInit):
         self.p.config.check_recovery_conf = Mock(return_value=(False, False))
         self.ha._leader_timeline = 1
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (leader) the standby leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (leader), the standby leader with the lock')
         self.p.set_role('replica')
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
@@ -766,7 +766,7 @@ class TestHa(PostgresInit):
         self.p.name = 'replica'
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
         self.assertEqual(self.ha.run_cycle(),
-                         'no action. I am a secondary (replica) and following a standby leader (leader)')
+                         'no action. I am (replica), a secondary, and following a standby leader (leader)')
         with patch.object(Leader, 'conn_url', PropertyMock(return_value='')):
             self.assertEqual(self.ha.run_cycle(), 'continue following the old known standby leader')
 
@@ -860,7 +860,7 @@ class TestHa(PostgresInit):
         self.ha.has_lock = false
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(),
-                         'no action. I am a secondary (postgresql0) and following a leader (leader)')
+                         'no action. I am (postgresql0), a secondary, and following a leader (leader)')
         check_calls([(update_lock, False), (demote, False)])
 
     def test_manual_failover_while_starting(self):
@@ -1085,7 +1085,7 @@ class TestHa(PostgresInit):
         self.ha.cluster.config.data.clear()
         self.ha.has_lock = true
         self.ha.cluster.is_unlocked = false
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
     def test_watch(self):
         self.ha.cluster = get_cluster_initialized_with_leader()
@@ -1120,7 +1120,7 @@ class TestHa(PostgresInit):
         self.ha.cluster.is_unlocked = false
         for tl in (1, 3):
             self.p.get_master_timeline = Mock(return_value=tl)
-            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
     @patch('sys.exit', return_value=1)
     def test_abort_join(self, exit_mock):
@@ -1133,9 +1133,9 @@ class TestHa(PostgresInit):
         self.ha.has_lock = true
         self.ha.cluster.is_unlocked = false
         self.ha.is_paused = true
-        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0), the leader with the lock')
         self.ha.is_paused = false
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
 
     @patch('psycopg2.connect', psycopg2_connect)
     def test_permanent_logical_slots_after_promote(self):
@@ -1145,7 +1145,7 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
         self.ha.cluster = get_cluster_initialized_without_leader(leader=True, cluster_config=config)
         self.ha.has_lock = true
-        self.assertEqual(self.ha.run_cycle(), 'no action. I am (other) the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (other), the leader with the lock')
 
     @patch.object(Cluster, 'has_member', true)
     def test_run_cycle(self):