Repository: zalando/patroni
Pull request #2583 (2023-03-03 10:30:14)

Don't write to PGDATA if major version is not known
It could happen that Patroni is started up before PGDATA was mounted. In this case Patroni can't determine major Postgres version from PG_VERSION file. Later, when PGDATA is mounted, Patroni was trying to create the recovery.conf even if the actual Postgres major version is newver than 12.

To mitigate the problem we double check that the `Postgresql._major_version` is set before writing recovery configuration or starting postgres up.

Close https://github.com/zalando/patroni/issues/2434

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

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

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/57504598/badge)](https://coveralls.io/builds/57504598) |
| :-- | --: |
| Change from base [Build 4312101654](https://coveralls.io/builds/57457984): |  7.0e-05% |
| Covered Lines: | 12099 |
| Relevant Lines: | 12117 |

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

hughcapet: 👍 

Patch:
diff --git a/patroni/ha.py b/patroni/ha.py
index 54a4667f..06294c3c 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -14,7 +14,7 @@ from threading import RLock
 from . import psycopg
 from .async_executor import AsyncExecutor, CriticalTask
 from .exceptions import DCSError, PostgresConnectionException, PatroniFatalException
-from .postgresql import ACTION_ON_START, ACTION_ON_ROLE_CHANGE
+from .postgresql.callback_executor import CallbackAction
 from .postgresql.misc import postgres_version_to_int
 from .postgresql.rewind import Rewind
 from .utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster, parse_int
@@ -567,7 +567,7 @@ class Ha(object):
                 self._rewind.trigger_check_diverged_lsn()
             elif role == 'standby_leader' and self.state_handler.role != role:
                 self.state_handler.set_role(role)
-                self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE)
+                self.state_handler.call_nowait(CallbackAction.ON_ROLE_CHANGE)
 
         return follow_reason
 
@@ -1502,7 +1502,7 @@ class Ha(object):
         self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
         self.dcs.take_leader()
         self.set_is_leader(True)
-        self.state_handler.call_nowait(ACTION_ON_START)
+        self.state_handler.call_nowait(CallbackAction.ON_START)
         self.load_cluster_from_dcs()
 
         return 'initialized a new cluster'
@@ -1700,7 +1700,7 @@ class Ha(object):
                 if not self.state_handler.cb_called:
                     if not self.state_handler.is_leader():
                         self._rewind.trigger_check_diverged_lsn()
-                    self.state_handler.call_nowait(ACTION_ON_START)
+                    self.state_handler.call_nowait(CallbackAction.ON_START)
                 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,
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index bed125c4..c8c7ef99 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -13,9 +13,10 @@ from datetime import datetime
 from dateutil import tz
 from psutil import TimeoutExpired
 from threading import current_thread, Lock
+from typing import Optional
 
 from .bootstrap import Bootstrap
-from .callback_executor import CallbackExecutor
+from .callback_executor import CallbackAction, CallbackExecutor
 from .cancellable import CancellableSubprocess
 from .config import ConfigHandler, mtime
 from .connection import Connection, get_connection_cursor
@@ -25,19 +26,13 @@ from .postmaster import PostmasterProcess
 from .slots import SlotsHandler
 from .sync import SyncHandler
 from .. import psycopg
+from ..dcs import Member
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
 
 
 logger = logging.getLogger(__name__)
 
-ACTION_ON_START = "on_start"
-ACTION_ON_STOP = "on_stop"
-ACTION_ON_RESTART = "on_restart"
-ACTION_ON_RELOAD = "on_reload"
-ACTION_ON_ROLE_CHANGE = "on_role_change"
-ACTION_NOOP = "noop"
-
 STATE_RUNNING = 'running'
 STATE_REJECT = 'rejecting connections'
 STATE_NO_RESPONSE = 'not responding'
@@ -207,7 +202,10 @@ class Postgresql(object):
     def _version_file_exists(self):
         return not self.data_directory_empty() and os.path.isfile(self._version_file)
 
-    def get_major_version(self):
+    def get_major_version(self) -> int:
+        """Reads major version from PG_VERSION file
+
+        :returns: major PostgreSQL version in integer format or 0 in case of missing file or errors"""
         if self._version_file_exists():
             try:
                 with open(self._version_file) as f:
@@ -496,21 +494,22 @@ class Postgresql(object):
     def cb_called(self):
         return self.__cb_called
 
-    def call_nowait(self, cb_name):
-        """ pick a callback command and call it without waiting for it to finish """
+    def call_nowait(self, cb_type: CallbackAction) -> None:
+        """pick a callback command and call it without waiting for it to finish """
         if self.bootstrapping:
             return
-        if cb_name in (ACTION_ON_START, ACTION_ON_STOP, ACTION_ON_RESTART, ACTION_ON_ROLE_CHANGE):
+        if cb_type in (CallbackAction.ON_START, CallbackAction.ON_STOP,
+                       CallbackAction.ON_RESTART, CallbackAction.ON_ROLE_CHANGE):
             self.__cb_called = True
 
-        if self.callback and cb_name in self.callback:
-            cmd = self.callback[cb_name]
+        if self.callback and cb_type in self.callback:
+            cmd = self.callback[cb_type]
             role = 'master' if self.role == 'promoted' else self.role
             try:
-                cmd = shlex.split(self.callback[cb_name]) + [cb_name, role, self.scope]
+                cmd = shlex.split(self.callback[cb_type]) + [cb_type, role, self.scope]
                 self._callback_executor.call(cmd)
             except Exception:
-                logger.exception('callback %s %s %s %s failed', cmd, cb_name, role, self.scope)
+                logger.exception('callback %s %r %s %s failed', cmd, cb_type, role, self.scope)
 
     @property
     def role(self):
@@ -563,7 +562,8 @@ class Postgresql(object):
         Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion
         or failure.
 
-        :returns: True if start was initiated and postmaster ports are open, False if start failed"""
+        :returns: True if start was initiated and postmaster ports are open,
+                  False if start failed, and None if postgres is still starting up"""
         # make sure we close all connections established against
         # the former node, otherwise, we might get a stalled one
         # after kill -9, which would report incorrect data to
@@ -576,7 +576,7 @@ class Postgresql(object):
             return True
 
         if not block_callbacks:
-            self.__cb_pending = ACTION_ON_START
+            self.__cb_pending = CallbackAction.ON_START
 
         self.set_role(role or self.get_postgres_role_from_data_directory())
 
@@ -584,8 +584,8 @@ class Postgresql(object):
         self._pending_restart = False
 
         try:
-            if not self._major_version:
-                self.configure_server_parameters()
+            if not self.ensure_major_version_is_known():
+                return None
             configuration = self.config.effective_configuration
         except Exception:
             return None
@@ -678,7 +678,7 @@ class Postgresql(object):
             if not block_callbacks:
                 self.set_state('stopped')
                 if pg_signaled:
-                    self.call_nowait(ACTION_ON_STOP)
+                    self.call_nowait(CallbackAction.ON_STOP)
         else:
             logger.warning('pg_ctl stop failed')
             self.set_state('stop failed')
@@ -770,7 +770,7 @@ class Postgresql(object):
     def reload(self, block_callbacks=False):
         ret = self.pg_ctl('reload')
         if ret and not block_callbacks:
-            self.call_nowait(ACTION_ON_RELOAD)
+            self.call_nowait(CallbackAction.ON_RELOAD)
         return ret
 
     def check_for_startup(self):
@@ -806,7 +806,7 @@ class Postgresql(object):
             self.config.save_configuration_files(True)
             # TODO: __cb_pending can be None here after PostgreSQL restarts on its own. Do we want to call the callback?
             # Previously we didn't even notice.
-            action = self.__cb_pending or ACTION_ON_START
+            action = self.__cb_pending or CallbackAction.ON_START
             self.call_nowait(action)
             self.__cb_pending = None
 
@@ -838,7 +838,7 @@ class Postgresql(object):
         """
         self.set_state('restarting')
         if not block_callbacks:
-            self.__cb_pending = ACTION_ON_RESTART
+            self.__cb_pending = CallbackAction.ON_RESTART
         ret = self.stop(block_callbacks=True, before_shutdown=before_shutdown)\
             and self.start(timeout, task, True, role, after_start)
         if not ret and not self.is_starting():
@@ -932,7 +932,27 @@ class Postgresql(object):
             except Exception:
                 logger.exception('Failed to read and parse %s', (history_path,))
 
-    def follow(self, member, role='replica', timeout=None, do_reload=False):
+    def follow(self,
+               member: Member,
+               role: Optional[str] = 'replica',
+               timeout: Optional[float] = None,
+               do_reload: Optional[bool] = False) -> Optional[bool]:
+        """Reconfigure postgres to follow a new member or use different recovery parameters.
+
+        Method may call `on_role_change` callback if role is changing.
+
+        :param member: The member to follow
+        :param role: The desired role, normally 'replica', but could also be a 'standby_leader'
+        :param timeout: start timeout, how long should the `start()` method wait for postgres accepting connections
+        :param do_reload: indicates that after updating postgresql.conf we just need to do a reload instead of restart
+
+        :returns: True - if restart/reload were successfully performed,
+                  False - if restart/reload failed
+                  None - if nothing was done or if Postgres is still in starting state after `timeout` seconds."""
+
+        if not self.ensure_major_version_is_known():
+            return None
+
         recovery_params = self.config.build_recovery_params(member)
         self.config.write_recovery_conf(recovery_params)
 
@@ -943,7 +963,7 @@ class Postgresql(object):
         change_role = self.cb_called and (self.role in ('master', 'primary', 'demoted') or
                                           not {'standby_leader', 'replica'} - {self.role, role})
         if change_role:
-            self.__cb_pending = ACTION_NOOP
+            self.__cb_pending = CallbackAction.NOOP
 
         ret = True
         if self.is_running():
@@ -959,7 +979,7 @@ class Postgresql(object):
 
         if change_role:
             # TODO: postpone this until start completes, or maybe do even earlier
-            self.call_nowait(ACTION_ON_ROLE_CHANGE)
+            self.call_nowait(CallbackAction.ON_ROLE_CHANGE)
         return ret
 
     def _wait_promote(self, wait_seconds):
@@ -1012,7 +1032,7 @@ class Postgresql(object):
             self.set_role('promoted')
             if on_success is not None:
                 on_success()
-            self.call_nowait(ACTION_ON_ROLE_CHANGE)
+            self.call_nowait(CallbackAction.ON_ROLE_CHANGE)
             ret = self._wait_promote(wait_seconds)
         return ret
 
@@ -1056,7 +1076,15 @@ class Postgresql(object):
     def configure_server_parameters(self):
         self._major_version = self.get_major_version()
         self.config.setup_server_parameters()
-        return True
+
+    def ensure_major_version_is_known(self) -> bool:
+        """Calls configure_server_parameters() if `_major_version` is not known
+
+        :returns: `True` if `_major_version` is set, otherwise `False`"""
+
+        if not self._major_version:
+            self.configure_server_parameters()
+        return self._major_version > 0
 
     def pg_wal_realpath(self):
         """Returns a dict containing the symlink (key) and target (value) for the wal directory"""
@@ -1149,8 +1177,7 @@ class Postgresql(object):
             2. sync replication slots, because it might happen that slots were removed
             3. get new 'Database system identifier' to make sure that it wasn't changed
         """
-        if not self._major_version:
-            self.configure_server_parameters()
+        self.ensure_major_version_is_known()
         self.slots_handler.schedule()
         self.citus_handler.schedule_cache_rebuild()
         self._sysid = None
diff --git a/patroni/postgresql/callback_executor.py b/patroni/postgresql/callback_executor.py
index 40ebf5a1..99711b08 100644
--- a/patroni/postgresql/callback_executor.py
+++ b/patroni/postgresql/callback_executor.py
@@ -1,22 +1,60 @@
 import logging
 
-from patroni.postgresql.cancellable import CancellableExecutor
+from enum import Enum
 from threading import Condition, Thread
+from typing import List
+
+from .cancellable import CancellableExecutor, CancellableSubprocess
 
 logger = logging.getLogger(__name__)
 
 
+class CallbackAction(str, Enum):
+    NOOP = "noop"
+    ON_START = "on_start"
+    ON_STOP = "on_stop"
+    ON_RESTART = "on_restart"
+    ON_RELOAD = "on_reload"
+    ON_ROLE_CHANGE = "on_role_change"
+
+    def __repr__(self):
+        return self.value
+
+
+class OnReloadExecutor(CancellableSubprocess):
+
+    def call_nowait(self, cmd: List[str]) -> None:
+        """Run one `on_reload` callback at most.
+
+        To achieve it we always kill already running command including child processes."""
+        self.cancel(kill=True)
+        self._kill_children()
+        with self._lock:
+            self._start_process(cmd, close_fds=True)
+
+
 class CallbackExecutor(CancellableExecutor, Thread):
 
     def __init__(self):
         CancellableExecutor.__init__(self)
         Thread.__init__(self)
         self.daemon = True
+        self._on_reload_executor = OnReloadExecutor()
         self._cmd = None
         self._condition = Condition()
         self.start()
 
-    def call(self, cmd):
+    def call(self, cmd: List[str]) -> None:
+        """Executes one callback at a time.
+
+        Already running command is killed (including child processes).
+        If it couldn't be killed we wait until it finishes.
+
+        :param cmd: command to be executed"""
+
+        if cmd[-3] == CallbackAction.ON_RELOAD:
+            return self._on_reload_executor.call_nowait(cmd)
+
         self._kill_process()
         with self._condition:
             self._cmd = cmd
diff --git a/patroni/postgresql/misc.py b/patroni/postgresql/misc.py
index 798bed35..462e5d43 100644
--- a/patroni/postgresql/misc.py
+++ b/patroni/postgresql/misc.py
@@ -7,7 +7,7 @@ from patroni.exceptions import PostgresException
 logger = logging.getLogger(__name__)
 
 
-def postgres_version_to_int(pg_version):
+def postgres_version_to_int(pg_version: str) -> int:
     """Convert the server_version to integer
 
     >>> postgres_version_to_int('9.5.3')
@@ -45,7 +45,7 @@ def postgres_version_to_int(pg_version):
     return int(''.join('{0:02d}'.format(c) for c in components))
 
 
-def postgres_major_version_to_int(pg_version):
+def postgres_major_version_to_int(pg_version: str) -> int:
     """
     >>> postgres_major_version_to_int('10')
     100000


Test patch:
diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py
index 9bbc8112..1c16306f 100644
--- a/tests/test_bootstrap.py
+++ b/tests/test_bootstrap.py
@@ -112,6 +112,7 @@ class TestBootstrap(BaseTestPostgresql):
         config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}}
 
         with patch.object(Postgresql, 'is_running', Mock(return_value=False)),\
+                patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)),\
                 patch('multiprocessing.Process', Mock(side_effect=Exception)),\
                 patch('multiprocessing.get_context', Mock(side_effect=Exception), create=True):
             self.assertRaises(Exception, self.b.bootstrap, config)
diff --git a/tests/test_callback_executor.py b/tests/test_callback_executor.py
index b52043ea..df2556b9 100644
--- a/tests/test_callback_executor.py
+++ b/tests/test_callback_executor.py
@@ -12,25 +12,28 @@ class TestCallbackExecutor(unittest.TestCase):
         mock_popen.return_value.children.return_value = []
         mock_popen.return_value.is_running.return_value = True
 
+        callback = ['test.sh', 'on_start', 'replica', 'foo']
         ce = CallbackExecutor()
         ce._kill_children = Mock(side_effect=Exception)
         ce._invoke_excepthook = Mock()
-        self.assertIsNone(ce.call([]))
+        self.assertIsNone(ce.call(callback))
         ce.join()
 
-        self.assertIsNone(ce.call([]))
+        self.assertIsNone(ce.call(callback))
 
         mock_popen.return_value.kill.side_effect = psutil.AccessDenied()
-        self.assertIsNone(ce.call([]))
+        self.assertIsNone(ce.call(callback))
 
         ce._process_children = []
         mock_popen.return_value.children.side_effect = psutil.Error()
         mock_popen.return_value.kill.side_effect = psutil.NoSuchProcess(123)
-        self.assertIsNone(ce.call([]))
+        self.assertIsNone(ce.call(callback))
 
         mock_popen.side_effect = Exception
         ce = CallbackExecutor()
         ce._condition.wait = Mock(side_effect=[None, Exception])
         ce._invoke_excepthook = Mock()
-        self.assertIsNone(ce.call([]))
+        self.assertIsNone(ce.call(callback))
+
+        self.assertIsNone(ce.call(['test.sh', 'on_reload', 'replica', 'foo']))
         ce.join()
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 65a3460c..10ae87af 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -182,6 +182,8 @@ def run_async(self, func, args=()):
 @patch.object(CancellableSubprocess, 'call', Mock(return_value=0))
 @patch.object(Postgresql, 'get_replica_timeline', Mock(return_value=2))
 @patch.object(Postgresql, 'get_primary_timeline', Mock(return_value=2))
+@patch.object(Postgresql, 'get_major_version', Mock(return_value=140000))
+@patch.object(Postgresql, 'resume_wal_replay', Mock())
 @patch.object(ConfigHandler, 'restore_configuration_files', Mock())
 @patch.object(etcd.Client, 'write', etcd_write)
 @patch.object(etcd.Client, 'read', etcd_read)
@@ -449,7 +451,6 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'not promoting because failed to update leader lock in DCS')
 
-    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000))
     def test_follow(self):
         self.ha.cluster.is_unlocked = false
         self.p.is_leader = false
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 20777a80..29dcbe93 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -14,6 +14,7 @@ from patroni.dcs import RemoteMember
 from patroni.exceptions import PostgresConnectionException, PatroniException
 from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
 from patroni.postgresql.bootstrap import Bootstrap
+from patroni.postgresql.callback_executor import CallbackAction
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.utils import RetryFailedError
 from six.moves import builtins
@@ -114,6 +115,9 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertTrue(self.p.start())
         mock_is_running.return_value = None
 
+        with patch.object(Postgresql, 'ensure_major_version_is_known', Mock(return_value=False)):
+            self.assertIsNone(self.p.start())
+
         mock_postmaster = MockPostmaster()
         with patch.object(PostmasterProcess, 'start', return_value=mock_postmaster):
             pg_conf = os.path.join(self.p.data_dir, 'postgresql.conf')
@@ -242,7 +246,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch('patroni.postgresql.config.mtime', mock_mtime)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test_check_recovery_conf(self, mock_get_pg_settings):
-        self.p.call_nowait('on_start')
+        self.p.call_nowait(CallbackAction.ON_START)
         mock_get_pg_settings.return_value = {
             'primary_conninfo': ['primary_conninfo', 'foo=', None, 'string', 'postmaster', self.p.config._auto_conf],
             'recovery_min_apply_delay': ['recovery_min_apply_delay', '0', 'ms', 'integer', 'sighup', 'foo']
@@ -278,7 +282,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(MockPostmaster, 'create_time', Mock(return_value=1234567), create=True)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test__read_recovery_params(self, mock_get_pg_settings):
-        self.p.call_nowait('on_start')
+        self.p.call_nowait(CallbackAction.ON_START)
         mock_get_pg_settings.return_value = {'primary_conninfo': ['primary_conninfo', '', None, 'string',
                                                                   'postmaster', self.p.config._postgresql_conf]}
         self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
@@ -320,9 +324,11 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_running', Mock(return_value=False))
     @patch.object(Postgresql, 'start', Mock())
     def test_follow(self):
-        self.p.call_nowait('on_start')
+        self.p.call_nowait(CallbackAction.ON_START)
         m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}})
         self.p.follow(m)
+        with patch.object(Postgresql, 'ensure_major_version_is_known', Mock(return_value=False)):
+            self.assertIsNone(self.p.follow(m))
 
     @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError))
     def test__query(self):
@@ -425,12 +431,9 @@ class TestPostgresql(BaseTestPostgresql):
     @patch('shlex.split', Mock(side_effect=OSError))
     def test_call_nowait(self):
         self.p.set_role('replica')
-        self.assertIsNone(self.p.call_nowait('on_start'))
+        self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START))
         self.p.bootstrapping = True
-        self.assertIsNone(self.p.call_nowait('on_start'))
-
-    def test_non_existing_callback(self):
-        self.assertFalse(self.p.call_nowait('foobar'))
+        self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START))
 
     @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
     def test_is_leader_exception(self):