Clean pg_replslot/ after pg_rewind (#2531)

As pg_rewind cleans this directory on target only since pg11

Co-authored-by: Alexander Kukushkin <[email protected]>

patroni/postgresql/misc.py

@@ -1,4 +1,6 @@
+import errno
 import logging
+import os
 
 from patroni.exceptions import PostgresException
 
@@ -73,3 +75,16 @@ def parse_history(data):
 def format_lsn(lsn, full=False):
     template = '{0:X}/{1:08X}' if full else '{0:X}/{1:X}'
     return template.format(lsn >> 32, lsn & 0xFFFFFFFF)
+
+
+def fsync_dir(path):
+    if os.name != 'nt':
+        fd = os.open(path, os.O_DIRECTORY)
+        try:
+            os.fsync(fd)
+        except OSError as e:
+            # Some filesystems don't like fsyncing directories and raise EINVAL. Ignoring it is usually safe.
+            if e.errno != errno.EINVAL:
+                raise
+        finally:
+            os.close(fd)

patroni/postgresql/rewind.py

@@ -9,7 +9,7 @@ import subprocess
 from threading import Lock, Thread
 
 from .connection import get_connection_cursor
-from .misc import format_lsn, parse_history, parse_lsn
+from .misc import format_lsn, fsync_dir, parse_history, parse_lsn
 from ..async_executor import CriticalTask
 from ..dcs import Leader
 
@@ -362,6 +362,18 @@ class Rewind(object):
                 else:
                     logger.info('Failed to archive WAL segment %s', wal)
 
+    def _maybe_clean_pg_replslot(self):
+        """Clean pg_replslot directory if pg version is less then 11
+        (pg_rewind deletes $PGDATA/pg_replslot content only since pg11)."""
+        if self._postgresql.major_version < 110000:
+            replslot_dir = self._postgresql.slots_handler.pg_replslot_dir
+            try:
+                for f in os.listdir(replslot_dir):
+                    shutil.rmtree(os.path.join(replslot_dir, f))
+                fsync_dir(replslot_dir)
+            except Exception as e:
+                logger.warning('Unable to clean %s: %r', replslot_dir, e)
+
     def pg_rewind(self, r):
         # prepare pg_rewind connection
         env = self._postgresql.config.write_pgpass(r)
@@ -439,6 +451,7 @@ class Rewind(object):
             return
 
         if self.pg_rewind(r):
+            self._maybe_clean_pg_replslot()
             self._state = REWIND_STATUS.SUCCESS
         else:
             if not self.check_leader_is_not_in_recovery(r):

patroni/postgresql/slots.py

@@ -1,4 +1,3 @@
-import errno
 import logging
 import os
 import shutil
@@ -8,7 +7,7 @@ from contextlib import contextmanager
 from threading import Condition, Thread
 
 from .connection import get_connection_cursor
-from .misc import format_lsn
+from .misc import format_lsn, fsync_dir
 from ..psycopg import OperationalError
 
 logger = logging.getLogger(__name__)
@@ -19,19 +18,6 @@ def compare_slots(s1, s2, dbid='database'):
                                          s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
 
 
-def fsync_dir(path):
-    if os.name != 'nt':
-        fd = os.open(path, os.O_DIRECTORY)
-        try:
-            os.fsync(fd)
-        except OSError as e:
-            # Some filesystems don't like fsyncing directories and raise EINVAL. Ignoring it is usually safe.
-            if e.errno != errno.EINVAL:
-                raise
-        finally:
-            os.close(fd)
-
-
 class SlotsAdvanceThread(Thread):
 
     def __init__(self, slots_handler):
@@ -122,6 +108,7 @@ class SlotsHandler(object):
         self._advance = None
         self._replication_slots = {}  # already existing replication slots
         self._unready_logical_slots = {}
+        self.pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
         self.schedule()
 
     def _query(self, sql, *params):
@@ -395,9 +382,8 @@ class SlotsHandler(object):
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
         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 create_slots.items():
-                slot_dir = os.path.join(pg_replslot_dir, name)
+                slot_dir = os.path.join(self._postgresql.slots_handler.pg_replslot_dir, name)
                 slot_tmp_dir = slot_dir + '.tmp'
                 if os.path.exists(slot_tmp_dir):
                     shutil.rmtree(slot_tmp_dir)
@@ -412,7 +398,7 @@ class SlotsHandler(object):
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
                 self._unready_logical_slots[name] = None
-            fsync_dir(pg_replslot_dir)
+            fsync_dir(self._postgresql.slots_handler.pg_replslot_dir)
             self._postgresql.start()
 
     def schedule(self, value=None):

tests/test_ha.py

@@ -310,6 +310,8 @@ class TestHa(PostgresInit):
 
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
+    @patch('os.listdir', Mock(return_value=[]))
+    @patch('patroni.postgresql.rewind.fsync_dir', Mock())
     def test_recover_with_rewind(self):
         self.p.is_running = false
         self.ha.cluster = get_cluster_initialized_with_leader()
@@ -817,6 +819,8 @@ class TestHa(PostgresInit):
 
     @patch.object(Rewind, 'pg_rewind', true)
     @patch.object(Rewind, 'check_leader_is_not_in_recovery', true)
+    @patch('os.listdir', Mock(return_value=[]))
+    @patch('patroni.postgresql.rewind.fsync_dir', Mock())
     def test_post_recover(self):
         self.p.is_running = false
         self.ha.has_lock = true

tests/test_rewind.py

@@ -274,6 +274,19 @@ class TestRewind(BaseTestPostgresql):
             self.r._archive_ready_wals()
             mock_logger_info.assert_not_called()
 
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=100000))
+    @patch('os.listdir', Mock(side_effect=[OSError, ['something', 'something_else']]))
+    @patch('shutil.rmtree', Mock())
+    @patch('patroni.postgresql.rewind.fsync_dir', Mock())
+    @patch('patroni.postgresql.rewind.logger.warning')
+    def test_maybe_clean_pg_replslot(self, mock_logger):
+        # failed to list pg_replslot/
+        self.assertIsNone(self.r._maybe_clean_pg_replslot())
+        mock_logger.assert_called_once()
+        mock_logger.reset_mock()
+
+        self.assertIsNone(self.r._maybe_clean_pg_replslot())
+
     @patch('os.unlink', Mock())
     @patch('os.listdir', Mock(return_value=[]))
     @patch('os.path.isfile', Mock(return_value=True))

tests/test_slots.py

@@ -9,7 +9,8 @@ from threading import Thread
 from patroni import psycopg
 from patroni.dcs import Cluster, ClusterConfig, Member
 from patroni.postgresql import Postgresql
-from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler, fsync_dir
+from patroni.postgresql.misc import fsync_dir
+from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler
 
 from . import BaseTestPostgresql, psycopg_connect, MockCursor