gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Don't allow on_reload callback kill other callbacks (#2578)

Browse Source 
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
This commit is contained in:
Alexander Kukushkin
2023-03-06 16:33:03 +01:00
committed by GitHub
parent 89595babdf
commit 09d0d78b74
5 changed files with 75 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/ha.py
+4 -4
View File
@@ -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,
patroni/postgresql/__init__.py
+17 -23
View File
@@ -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
patroni/postgresql/callback_executor.py
+40 -2
View File
@@ -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
tests/test_callback_executor.py
+8 -5
View File
@@ -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()
tests/test_postgresql.py
+6 -8
View File
@@ -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):