Repository: zalando/patroni
Pull request #2382 (2022-08-15 08:10:20)

Apply timeout when waiting for user backends to close
Close https://github.com/zalando/patroni/issues/2365

coveralls: ## Pull Request Test Coverage Report for [Build 2859363496](https://coveralls.io/builds/51655722)

* **10** of **10**   **(100.0%)**  changed or added relevant lines in **2** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+3.0e**-05%**) to **99.882%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/51655722/badge)](https://coveralls.io/builds/51655722) |
| :-- | --: |
| Change from base [Build 2797700302](https://coveralls.io/builds/51422778): |  3.0e-05% |
| Covered Lines: | 11012 |
| Relevant Lines: | 11025 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/docs/pause.rst b/docs/pause.rst
index a33b04b4..285ebac1 100644
--- a/docs/pause.rst
+++ b/docs/pause.rst
@@ -27,6 +27,8 @@ When Patroni runs in a paused mode, it does not change the state of PostgreSQL,
 
 - When Postgres is stopped, Patroni does not try to start it. When Patroni is stopped, it does not try to stop the Postgres instance it is managing.
 
+- Patroni will not try to remove replication slots that don't represent the other cluster member or are not listed in the configuration of the permanent slots.
+
 User guide
 ----------
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 9f84613f..45cf538d 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -1407,7 +1407,14 @@ class Ha(object):
                     return 'started as a secondary'
 
             # is data directory empty?
-            if self.state_handler.data_directory_empty():
+            try:
+                data_directory_is_empty = self.state_handler.data_directory_empty()
+                data_directory_is_accessible = True
+            except OSError as e:
+                data_directory_is_accessible = False
+                data_directory_error = e
+
+            if not data_directory_is_accessible or data_directory_is_empty:
                 self.state_handler.set_role('uninitialized')
                 self.state_handler.stop('immediate', stop_timeout=self.patroni.config['retry_timeout'])
                 # In case datadir went away while we were master.
@@ -1416,8 +1423,11 @@ class Ha(object):
                 # is this instance the leader?
                 if self.has_lock():
                     self.release_leader_key_voluntarily()
-                    return 'released leader key voluntarily as data dir empty and currently leader'
+                    return 'released leader key voluntarily as data dir {0} and currently leader'.format(
+                                'empty' if data_directory_is_accessible else 'not accessible')
 
+                if not data_directory_is_accessible:
+                    return 'data directory is not accessible: {0}'.format(data_directory_error)
                 if self.is_paused():
                     return 'running with empty data directory'
                 return self.bootstrap()  # new node
@@ -1483,7 +1493,9 @@ class Ha(object):
                 # asynchronous processes are running (should be always the case for the master)
                 if not self._async_executor.busy and not self.state_handler.is_starting():
                     create_slots = self.state_handler.slots_handler.sync_replication_slots(self.cluster,
-                                                                                           self.patroni.nofailover)
+                                                                                           self.patroni.nofailover,
+                                                                                           self.patroni.replicatefrom,
+                                                                                           self.is_paused())
                     if not self.state_handler.cb_called:
                         if not self.state_handler.is_leader():
                             self._rewind.trigger_check_diverged_lsn()
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index a5b3d592..3e2e1d38 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -659,7 +659,7 @@ class Postgresql(object):
         if on_safepoint:
             # Wait for our connection to terminate so we can be sure that no new connections are being initiated
             self._wait_for_connection_close(postmaster)
-            postmaster.wait_for_user_backends_to_close()
+            postmaster.wait_for_user_backends_to_close(stop_timeout)
             on_safepoint()
 
         if on_shutdown and mode in ('fast', 'smart'):
diff --git a/patroni/postgresql/postmaster.py b/patroni/postgresql/postmaster.py
index f7b39622..e2a88856 100644
--- a/patroni/postgresql/postmaster.py
+++ b/patroni/postgresql/postmaster.py
@@ -171,8 +171,8 @@ class PostmasterProcess(psutil.Process):
         else:
             return not self.is_running()
 
-    def wait_for_user_backends_to_close(self):
-        # These regexps are cross checked against versions PostgreSQL 9.1 .. 11
+    def wait_for_user_backends_to_close(self, stop_timeout):
+        # These regexps are cross checked against versions PostgreSQL 9.1 .. 15
         aux_proc_re = re.compile("(?:postgres:)( .*:)? (?:(?:archiver|startup|autovacuum launcher|autovacuum worker|"
                                  "checkpointer|logger|stats collector|wal receiver|wal writer|writer)(?: process  )?|"
                                  "walreceiver|wal sender process|walsender|walwriter|background writer|"
@@ -184,19 +184,23 @@ class PostmasterProcess(psutil.Process):
             return logger.debug('Failed to get list of postmaster children')
 
         user_backends = []
-        user_backends_cmdlines = []
+        user_backends_cmdlines = {}
         for child in children:
             try:
                 cmdline = child.cmdline()
                 if cmdline and not aux_proc_re.match(cmdline[0]):
                     user_backends.append(child)
-                    user_backends_cmdlines.append(cmdline[0])
+                    user_backends_cmdlines[child.pid] = cmdline[0]
             except psutil.NoSuchProcess:
                 pass
         if user_backends:
-            logger.debug('Waiting for user backends %s to close', ', '.join(user_backends_cmdlines))
-            psutil.wait_procs(user_backends)
-        logger.debug("Backends closed")
+            logger.debug('Waiting for user backends %s to close', ', '.join(user_backends_cmdlines.values()))
+            gone, live = psutil.wait_procs(user_backends, stop_timeout)
+            if stop_timeout and live:
+                live = [user_backends_cmdlines[b.pid] for b in live]
+                logger.warning('Backends still alive after %s: %s', stop_timeout, ', '.join(live))
+            else:
+                logger.debug("Backends closed")
 
     @staticmethod
     def start(pgcommand, data_dir, conf, options):
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index 32cbc9b8..0d5051f7 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -112,10 +112,10 @@ class SlotsHandler(object):
         # In normal situation rowcount should be 1, otherwise either slot doesn't exists or it is still active
         return cursor.rowcount == 1
 
-    def _drop_incorrect_slots(self, cluster, slots):
+    def _drop_incorrect_slots(self, cluster, slots, paused):
         # drop old replication slots which are not presented in desired slots
         for name in set(self._replication_slots) - set(slots):
-            if not self.ignore_replication_slot(cluster, name) and not self.drop_replication_slot(name):
+            if not paused and 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
 
@@ -206,7 +206,7 @@ class SlotsHandler(object):
                         self._schedule_load_slots = True
         return create_slots
 
-    def sync_replication_slots(self, cluster, nofailover, replicatefrom=None):
+    def sync_replication_slots(self, cluster, nofailover, replicatefrom=None, paused=False):
         ret = None
         if self._postgresql.major_version >= 90400 and cluster.config:
             try:
@@ -215,7 +215,7 @@ class SlotsHandler(object):
                 slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role,
                                                       nofailover, self._postgresql.major_version, True)
 
-                self._drop_incorrect_slots(cluster, slots)
+                self._drop_incorrect_slots(cluster, slots, paused)
 
                 self._ensure_physical_slots(slots)
 


Test patch:
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 5ed4503d..1d6e2904 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -1133,17 +1133,19 @@ class TestHa(PostgresInit):
         self.ha.shutdown()
 
     @patch('time.sleep', Mock())
-    def test_leader_with_empty_directory(self):
+    def test_leader_with_not_accessible_data_directory(self):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.ha.has_lock = true
-        self.p.data_directory_empty = true
-        self.assertEqual(self.ha.run_cycle(), 'released leader key voluntarily as data dir empty and currently leader')
+        self.p.data_directory_empty = Mock(side_effect=OSError(5, "Input/output error: '{}'".format(self.p.data_dir)))
+        self.assertEqual(self.ha.run_cycle(),
+                         'released leader key voluntarily as data dir not accessible and currently leader')
         self.assertEqual(self.p.role, 'uninitialized')
 
         # as has_lock is mocked out, we need to fake the leader key release
         self.ha.has_lock = false
-        # will not say bootstrap from leader as replica can't self elect
-        self.assertEqual(self.ha.run_cycle(), "trying to bootstrap from replica 'other'")
+        # will not say bootstrap because data directory is not accessible
+        self.assertEqual(self.ha.run_cycle(),
+                         "data directory is not accessible: [Errno 5] Input/output error: '{}'".format(self.p.data_dir))
 
     @patch('patroni.postgresql.mtime', Mock(return_value=1588316884))
     @patch.object(builtins, 'open', mock_open(read_data=('1\t0/40159C0\tno recovery target specified\n\n'
diff --git a/tests/test_postmaster.py b/tests/test_postmaster.py
index 07f09de2..61a45049 100644
--- a/tests/test_postmaster.py
+++ b/tests/test_postmaster.py
@@ -133,14 +133,20 @@ class TestPostmasterProcess(unittest.TestCase):
         c2.cmdline = Mock(return_value=["postgres: postgres postgres [local] idle"])
         c3 = Mock()
         c3.cmdline = Mock(side_effect=psutil.NoSuchProcess(123))
+        mock_wait.return_value = ([], [c2])
         with patch('psutil.Process.children', Mock(return_value=[c1, c2, c3])):
             proc = PostmasterProcess(123)
-            self.assertIsNone(proc.wait_for_user_backends_to_close())
-            mock_wait.assert_called_with([c2])
+            self.assertIsNone(proc.wait_for_user_backends_to_close(1))
+            mock_wait.assert_called_with([c2], 1)
+
+        mock_wait.return_value = ([c2], [])
+        with patch('psutil.Process.children', Mock(return_value=[c1, c2, c3])):
+            proc = PostmasterProcess(123)
+            proc.wait_for_user_backends_to_close(1)
 
         with patch('psutil.Process.children', Mock(side_effect=psutil.NoSuchProcess(123))):
             proc = PostmasterProcess(123)
-            self.assertIsNone(proc.wait_for_user_backends_to_close())
+            self.assertIsNone(proc.wait_for_user_backends_to_close(None))
 
     @patch('subprocess.Popen')
     @patch('os.setsid', Mock(), create=True)
diff --git a/tests/test_slots.py b/tests/test_slots.py
index 454fab8d..2301ae5f 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -42,8 +42,10 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.set_role('standby_leader')
         self.s.sync_replication_slots(cluster, False)
         self.p.set_role('replica')
-        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)):
-            self.s.sync_replication_slots(cluster, False)
+        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)),\
+                patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
+            self.s.sync_replication_slots(cluster, False, paused=True)
+            mock_drop.assert_not_called()
         self.p.set_role('master')
         with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
             self.s.sync_replication_slots(cluster, False)