From 09d0d78b749fc6b79522836275d3f6ca24d3ecf2 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 6 Mar 2023 16:33:03 +0100 Subject: [PATCH] Don't allow on_reload callback kill other callbacks (#2578) Since a long time Patroni enforcing only one callback script running at a time. If the new callback is executed while the old one is still running, the old one is killed (including all child processes). Such behavior is fine for all callbacks but on_reload, because the last one may accidentally cancel important ones, that for example updating DNS or assigning/removing Virtual IP. To mitigate the problem we introduce a dedicated executor for on_reload callbacks, so that on_reload may only cancel another on_reload. Ref: https://github.com/zalando/patroni/issues/2445 --- patroni/ha.py | 8 ++--- patroni/postgresql/__init__.py | 40 ++++++++++------------- patroni/postgresql/callback_executor.py | 42 +++++++++++++++++++++++-- tests/test_callback_executor.py | 13 +++++--- tests/test_postgresql.py | 14 ++++----- 5 files changed, 75 insertions(+), 42 deletions(-) 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..16718108 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -15,7 +15,7 @@ from psutil import TimeoutExpired from threading import current_thread, Lock 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 @@ -31,13 +31,6 @@ from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_emp 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' @@ -496,21 +489,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): @@ -576,7 +570,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()) @@ -678,7 +672,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 +764,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 +800,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 +832,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(): @@ -943,7 +937,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 +953,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 +1006,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 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/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_postgresql.py b/tests/test_postgresql.py index 20777a80..71f5df4d 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 @@ -242,7 +243,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 +279,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,7 +321,7 @@ 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) @@ -425,12 +426,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):