Repository: zalando/patroni
Pull request #2578 (2023-03-01 15:22:34)

Don't allow on_reload callback kill other callbacks
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

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

* **0** of **18**   **(100.0%)**  changed or added relevant lines in **2** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.0001%**) to **99.851%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/57425473/badge)](https://coveralls.io/builds/57425473) |
| :-- | --: |
| Change from base [Build 4291292444](https://coveralls.io/builds/57361885): |  0.0001% |
| Covered Lines: | 12102 |
| Relevant Lines: | 12120 |

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


Patch:
diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index 88aa6ca5..be8249fb 100644
--- a/docs/rest_api.rst
+++ b/docs/rest_api.rst
@@ -111,6 +111,66 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be
       }
     }
 
+
+Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` endpoint.
+
+.. code-block:: bash
+
+	$ curl http://localhost:8008/metrics
+	
+	# HELP patroni_version Patroni semver without periods. \
+	# TYPE patroni_version gauge
+	patroni_version{scope="batman"} 020103
+	# HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise.
+	# TYPE patroni_postgres_running gauge
+	patroni_postgres_running{scope="batman"} 1
+	# HELP patroni_postmaster_start_time Epoch seconds since Postgres started.
+	# TYPE patroni_postmaster_start_time gauge
+	patroni_postmaster_start_time{scope="batman"} 1657656955.179243
+	# HELP patroni_master Value is 1 if this node is the leader, 0 otherwise.
+	# TYPE patroni_master gauge
+	patroni_master{scope="batman"} 1
+	# HELP patroni_xlog_location Current location of the Postgres transaction log, 0 if this node is not the leader.
+	# TYPE patroni_xlog_location counter
+	patroni_xlog_location{scope="batman"} 22320573386952
+	# HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise.
+	# TYPE patroni_standby_leader gauge
+	patroni_standby_leader{scope="batman"} 0
+	# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.
+	# TYPE patroni_replica gauge
+	patroni_replica{scope="batman"} 0
+	# HELP patroni_xlog_received_location Current location of the received Postgres transaction log, 0 if this node is not a replica.
+	# TYPE patroni_xlog_received_location counter
+	patroni_xlog_received_location{scope="batman"} 0
+	# HELP patroni_xlog_replayed_location Current location of the replayed Postgres transaction log, 0 if this node is not a replica.
+	# TYPE patroni_xlog_replayed_location counter
+	patroni_xlog_replayed_location{scope="batman"} 0
+	# HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed Postgres transaction log, 0 if null.
+	# TYPE patroni_xlog_replayed_timestamp gauge
+	patroni_xlog_replayed_timestamp{scope="batman"} 0
+	# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.
+	# TYPE patroni_xlog_paused gauge
+	patroni_xlog_paused{scope="batman"} 0
+	# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.
+	# TYPE patroni_postgres_server_version gauge
+	patroni_postgres_server_version {scope="batman"} 140004
+	# HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked.
+	# TYPE patroni_cluster_unlocked gauge
+	patroni_cluster_unlocked{scope="batman"} 0
+	# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.
+	# TYPE patroni_postgres_timeline counter
+	patroni_postgres_timeline{scope="batman"} 24
+	# HELP patroni_dcs_last_seen Epoch timestamp when DCS was last contacted successfully by Patroni.
+	# TYPE patroni_dcs_last_seen gauge
+	patroni_dcs_last_seen{scope="batman"} 1677658321
+	# HELP patroni_pending_restart Value is 1 if the node needs a restart, 0 otherwise.
+	# TYPE patroni_pending_restart gauge
+	patroni_pending_restart{scope="batman"} 1
+	# HELP patroni_is_paused Value is 1 if auto failover is disabled, 0 otherwise.
+	# TYPE patroni_is_paused gauge
+	patroni_is_paused{scope="batman"} 1
+
+
 Cluster status endpoints
 ------------------------
 
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/setup.py b/setup.py
index 9ebdbe21..1c0d3b3b 100644
--- a/setup.py
+++ b/setup.py
@@ -94,7 +94,7 @@ class Flake8(_Command):
         from flake8.main.cli import main
 
         logging.getLogger().setLevel(logging.ERROR)
-        main(self.targets())
+        raise SystemExit(main(self.targets()))
 
 
 class PyTest(_Command):


Test patch:
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):