Serialize callback execution (#366)

If the previous callback is still running - kill it
Also it will fix a problem of zombie processes when executing callbacks from the main thread.

patroni/callback_executor.py

@@ -0,0 +1,38 @@
+import logging
+import os
+import subprocess
+from threading import Event, Lock, Thread
+
+logger = logging.getLogger(__name__)
+
+
+class CallbackExecutor(Thread):
+
+    def __init__(self):
+        super(CallbackExecutor, self).__init__()
+        self.daemon = True
+        self._lock = Lock()
+        self._cmd = None
+        self._process = None
+        self._callback_event = Event()
+        self.start()
+
+    def call(self, cmd):
+        with self._lock:
+            if self._process and self._process.poll() is None:
+                self._process.kill()
+                logger.warning('Killed the old callback process because it was still running: %s', self._cmd)
+        self._cmd = cmd
+        self._callback_event.set()
+
+    def run(self):
+        while True:
+            self._callback_event.wait()
+            self._callback_event.clear()
+            with self._lock:
+                try:
+                    self._process = subprocess.Popen(self._cmd, close_fds=True, env={'PATH': os.environ.get('PATH')})
+                except Exception:
+                    logger.exception('Failed to execute %s',  self._cmd)
+                    continue
+            self._process.wait()

patroni/postgresql.py

@@ -10,6 +10,7 @@ import time
 
 from collections import defaultdict
 from patroni import call_self
+from patroni.callback_executor import CallbackExecutor
 from patroni.exceptions import PostgresConnectionException, PostgresException
 from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop
 from six import string_types
@@ -103,6 +104,7 @@ class Postgresql(object):
         self._schedule_load_slots = self.use_slots
 
         self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
+        self._callback_executor = CallbackExecutor()
         self.__cb_called = False
         self.__cb_pending = None
         config_base_name = config.get('config_base_name', 'postgresql')
@@ -607,15 +609,8 @@ class Postgresql(object):
         if cb_name in (ACTION_ON_START, ACTION_ON_STOP, ACTION_ON_RESTART, ACTION_ON_ROLE_CHANGE):
             self.__cb_called = True
 
-        if not self.callback or cb_name not in self.callback:
-            return False
-        cmd = self.callback[cb_name]
-        try:
-            subprocess.Popen(shlex.split(cmd) + [cb_name, self.role, self.scope])
-        except OSError:
-            logger.exception('callback %s %s %s %s failed', cmd, cb_name, self.role, self.scope)
-            return False
-        return True
+        if self.callback and cb_name in self.callback:
+            self._callback_executor.call(self.callback[cb_name])
 
     @property
     def role(self):

tests/test_callback_executor.py

@@ -0,0 +1,24 @@
+import unittest
+
+from mock import Mock, patch
+from patroni.callback_executor import CallbackExecutor
+
+
+class TestCallbackExecutor(unittest.TestCase):
+
+    @patch('subprocess.Popen')
+    def test_callback_executor(self, mock_popen):
+        mock_popen.return_value.wait.side_effect = Exception
+        mock_popen.return_value.poll.return_value = None
+
+        ce = CallbackExecutor()
+        self.assertIsNone(ce.call([]))
+        ce.join()
+
+        self.assertIsNone(ce.call([]))
+
+        mock_popen.side_effect = Exception
+        ce = CallbackExecutor()
+        ce._callback_event.wait = Mock(side_effect=[None, Exception])
+        self.assertIsNone(ce.call([]))
+        ce.join()

tests/test_ha.py

@@ -119,6 +119,7 @@ def run_async(self, func, args=()):
 @patch.object(Postgresql, 'write_recovery_conf', Mock())
 @patch.object(Postgresql, 'query', Mock())
 @patch.object(Postgresql, 'checkpoint', Mock())
+@patch.object(Postgresql, 'call_nowait', Mock())
 @patch.object(etcd.Client, 'write', etcd_write)
 @patch.object(etcd.Client, 'read', etcd_read)
 @patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException))

tests/test_patroni.py

@@ -27,6 +27,7 @@ class MockFrozenImporter(object):
 @patch.object(Postgresql, '_write_postgresql_conf', Mock())
 @patch.object(Postgresql, 'write_recovery_conf', Mock())
 @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+@patch.object(Postgresql, 'call_nowait', Mock())
 @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
 @patch.object(AsyncExecutor, 'run', Mock())
 @patch.object(etcd.Client, 'write', etcd_write)

tests/test_postgresql.py

@@ -184,6 +184,7 @@ class TestPostgresql(unittest.TestCase):
                                            'on_reload': 'true'
                                            },
                              'restore': 'true'})
+        self.p._callback_executor = Mock()
         self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
         self.leader = Leader(-1, 28, self.leadermem)
         self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',