From 838653325a2bfa2080d3ba11fced1bd1921ae4bb Mon Sep 17 00:00:00 2001 From: Polina Bungina Date: Thu, 19 Jan 2023 15:50:30 +0100 Subject: [PATCH] Clean pg_replslot/ after pg_rewind (#2531) As pg_rewind cleans this directory on target only since pg11 Co-authored-by: Alexander Kukushkin --- patroni/postgresql/misc.py | 15 +++++++++++++++ patroni/postgresql/rewind.py | 15 ++++++++++++++- patroni/postgresql/slots.py | 22 ++++------------------ tests/test_ha.py | 4 ++++ tests/test_rewind.py | 13 +++++++++++++ tests/test_slots.py | 3 ++- 6 files changed, 52 insertions(+), 20 deletions(-) diff --git a/patroni/postgresql/misc.py b/patroni/postgresql/misc.py index 3a6faa1c..798bed35 100644 --- a/patroni/postgresql/misc.py +++ b/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) diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 595105ba..76e6a54f 100644 --- a/patroni/postgresql/rewind.py +++ b/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): diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 7abe2182..00f8ac76 100644 --- a/patroni/postgresql/slots.py +++ b/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): diff --git a/tests/test_ha.py b/tests/test_ha.py index c9c2f129..04d2a0b9 100644 --- a/tests/test_ha.py +++ b/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 diff --git a/tests/test_rewind.py b/tests/test_rewind.py index b9398205..9da500e9 100644 --- a/tests/test_rewind.py +++ b/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)) diff --git a/tests/test_slots.py b/tests/test_slots.py index 4b19fc4a..7ee5a7a1 100644 --- a/tests/test_slots.py +++ b/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