Solve issue of handling sigchld when dunning in a docker (#355)

If Patroni was started in a docker with pid=1 it will execute itself
with the same arguments. The original process will take care about init
process duties, i.e. handle sigchld and reap dead orphan processes.
Also it will forward SIGINT, SIGHUP, SIGTERM and some other signals to
the real Patroni process.

patroni/__init__.py

@@ -1,23 +1,22 @@
 import logging
+import os
 import signal
 import sys
 import time
 
-from patroni.api import RestApiServer
-from patroni.config import Config
-from patroni.dcs import get_dcs
-from patroni.exceptions import DCSError
-from patroni.ha import Ha
-from patroni.postgresql import Postgresql
-from patroni.utils import reap_children, sigchld_handler
-from patroni.version import __version__
-
 logger = logging.getLogger(__name__)
 
 
 class Patroni(object):
 
     def __init__(self):
+        from patroni.api import RestApiServer
+        from patroni.config import Config
+        from patroni.dcs import get_dcs
+        from patroni.ha import Ha
+        from patroni.postgresql import Postgresql
+        from patroni.version import __version__
+
         self.setup_signal_handlers()
 
         self.version = __version__
@@ -34,6 +33,7 @@ class Patroni(object):
         self.scheduled_restart = {}
 
     def load_dynamic_configuration(self):
+        from patroni.exceptions import DCSError
         while True:
             try:
                 cluster = self.dcs.get_cluster()
@@ -107,8 +107,6 @@ class Patroni(object):
                 if self.config.reload_local_configuration():
                     self.reload_config()
 
-            reap_children()
-
             logger.info(self.ha.run_cycle())
 
             cluster = self.dcs.cluster
@@ -118,7 +116,6 @@ class Patroni(object):
             if not self.postgresql.data_directory_empty():
                 self.config.save_cache()
 
-            reap_children()
             self.schedule_next_run()
 
     def setup_signal_handlers(self):
@@ -126,10 +123,9 @@ class Patroni(object):
         self._received_sigterm = False
         signal.signal(signal.SIGHUP, self.sighup_handler)
         signal.signal(signal.SIGTERM, self.sigterm_handler)
-        signal.signal(signal.SIGCHLD, sigchld_handler)
 
 
-def main():
+def patroni_main():
     logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
     logging.getLogger('requests').setLevel(logging.WARNING)
 
@@ -145,3 +141,39 @@ def main():
         else:
             patroni.ha.while_not_sync_standby(lambda: patroni.postgresql.stop(checkpoint=False))
             patroni.dcs.delete_leader()
+
+
+def main():
+    if os.getpid() != 1:
+        return patroni_main()
+
+    pid = 0
+
+    # Looks like we are in a docker, so we will act like init
+    def sigchld_handler(signo, stack_frame):
+        try:
+            while True:
+                ret = os.waitpid(-1, os.WNOHANG)
+                if ret == (0, 0):
+                    break
+                elif ret[0] != pid:
+                    logging.info('Reaped pid=%s, exit status=%s', *ret)
+        except OSError:
+            pass
+
+    def passtochild(signo, stack_frame):
+        if pid:
+            os.kill(pid, signo)
+
+    signal.signal(signal.SIGCHLD, sigchld_handler)
+    signal.signal(signal.SIGHUP, passtochild)
+    signal.signal(signal.SIGINT, passtochild)
+    signal.signal(signal.SIGUSR1, passtochild)
+    signal.signal(signal.SIGUSR2, passtochild)
+    signal.signal(signal.SIGQUIT, passtochild)
+    signal.signal(signal.SIGTERM, passtochild)
+
+    import subprocess
+    patroni = subprocess.Popen([sys.executable] + sys.argv)
+    pid = patroni.pid
+    patroni.wait()

patroni/dcs/consul.py

@@ -8,7 +8,7 @@ import urllib3
 from consul import ConsulException, NotFound, base
 from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState
 from patroni.exceptions import DCSError
-from patroni.utils import Retry, RetryFailedError, sleep
+from patroni.utils import Retry, RetryFailedError
 from urllib3.exceptions import HTTPError
 from six.moves.urllib.parse import urlencode
 from six.moves.http_client import HTTPException
@@ -121,7 +121,7 @@ class Consul(AbstractDCS):
                 self.refresh_session()
             except ConsulError:
                 logger.info('waiting on consul')
-                sleep(5)
+                time.sleep(5)
 
     def set_ttl(self, ttl):
         if self._client.http.set_ttl(ttl/2.0):  # Consul multiplies the TTL by 2x

patroni/dcs/etcd.py

@@ -11,7 +11,7 @@ from dns.exception import DNSException
 from dns import resolver
 from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState
 from patroni.exceptions import DCSError
-from patroni.utils import Retry, RetryFailedError, sleep
+from patroni.utils import Retry, RetryFailedError
 from urllib3.exceptions import HTTPError, ReadTimeoutError
 from requests.exceptions import RequestException
 from six.moves.http_client import HTTPException
@@ -251,7 +251,7 @@ class Etcd(AbstractDCS):
                 client = Client(config)
             except etcd.EtcdException:
                 logger.info('waiting on etcd')
-                sleep(5)
+                time.sleep(5)
         return client
 
     def set_ttl(self, ttl):

patroni/dcs/exhibitor.py

@@ -4,7 +4,6 @@ import requests
 import time
 
 from patroni.dcs.zookeeper import ZooKeeper
-from patroni.utils import sleep
 from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)
@@ -24,7 +23,7 @@ class ExhibitorEnsembleProvider(object):
         self._next_poll = None
         while not self.poll():
             logger.info('waiting on exhibitor')
-            sleep(5)
+            time.sleep(5)
 
     def poll(self):
         if self._next_poll and self._next_poll > time.time():

patroni/ha.py

@@ -1,17 +1,18 @@
+import datetime
 import functools
 import json
 import logging
 import psycopg2
 import requests
 import sys
-import datetime
+import time
-from threading import RLock
 
 from multiprocessing.pool import ThreadPool
 from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
 from patroni.postgresql import ACTION_ON_START
-from patroni.utils import polling_loop, sleep, tzutc
+from patroni.utils import polling_loop, tzutc
+from threading import RLock
 
 logger = logging.getLogger(__name__)
 
@@ -202,7 +203,7 @@ class Ha(object):
 
                 if picked and not allow_promote:
                     # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby
-                    sleep(2)
+                    time.sleep(2)
                     picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster)
                 if allow_promote:
                     cluster = self.dcs.get_cluster()
@@ -429,7 +430,7 @@ class Ha(object):
             self.state_handler.set_role('demoted')
             self.dcs.delete_leader()
             self.dcs.reset_cluster()
-            sleep(2)  # Give a time to somebody to take the leader lock
+            time.sleep(2)  # Give a time to somebody to take the leader lock
             cluster = self.dcs.get_cluster()
             node_to_follow = self._get_node_to_follow(cluster)
             return self.state_handler.follow(node_to_follow, cluster.leader, recovery=True, need_rewind=True)
@@ -459,7 +460,7 @@ class Ha(object):
                     return False
 
                 # The value is very close to now
-                sleep(max(delta, 0))
+                time.sleep(max(delta, 0))
                 logger.info('Manual scheduled {0} at %s'.format(action_name), scheduled_at.isoformat())
                 return True
             except TypeError:
@@ -528,7 +529,7 @@ class Ha(object):
             # node tagged as nofailover can be ahead of the new leader either, but it is always excluded from elections
             need_rewind = bool(self.cluster.failover) or self.patroni.nofailover
             if need_rewind:
-                sleep(2)  # Give a time to somebody to take the leader lock
+                time.sleep(2)  # Give a time to somebody to take the leader lock
 
             if self.patroni.nofailover:
                 return self.follow('demoting self because I am not allowed to become master',

patroni/utils.py

@@ -1,4 +1,3 @@
-import os
 import random
 import sys
 import time
@@ -11,8 +10,6 @@ if sys.hexversion >= 0x3000000:
     long = int
 
 tzutc = tz.tzutc()
-__interrupted_sleep = False
-__reap_children = False
 
 
 def deep_compare(obj1, obj2):
@@ -197,36 +194,8 @@ def compare_values(vartype, unit, old_value, new_value):
     return old_value is not None and new_value is not None and old_value == new_value
 
 
-def sigchld_handler(signo, stack_frame):
+def _sleep(interval):
-    global __interrupted_sleep, __reap_children
+    time.sleep(interval)
-    __reap_children = __interrupted_sleep = True
-
-
-def sleep(interval):
-    global __interrupted_sleep
-    current_time = time.time()
-    end_time = current_time + interval
-    while current_time < end_time:
-        __interrupted_sleep = False
-        time.sleep(end_time - current_time)
-        if not __interrupted_sleep:  # we will ignore only sigchld
-            break
-        current_time = time.time()
-    __interrupted_sleep = False
-
-
-def reap_children():
-    global __reap_children
-    if __reap_children:
-        try:
-            while True:
-                ret = os.waitpid(-1, os.WNOHANG)
-                if ret == (0, 0):
-                    break
-        except OSError:
-            pass
-        finally:
-            __reap_children = False
 
 
 def is_valid_pg_version(version):
@@ -243,7 +212,7 @@ class Retry(object):
     """Helper for retrying a method in the face of retry-able exceptions"""
 
     def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.8, max_delay=3600,
-                 sleep_func=sleep, deadline=None, retry_exceptions=PatroniException):
+                 sleep_func=_sleep, deadline=None, retry_exceptions=PatroniException):
         """Create a :class:`Retry` instance for retrying function calls
 
         :param max_tries: How many times to retry the command. -1 means infinite tries.
@@ -314,4 +283,4 @@ def polling_loop(timeout, interval=1):
     while time.time() < end_time:
         yield iteration
         iteration += 1
-        sleep(interval)
+        time.sleep(interval)

tests/test_ha.py

@@ -514,7 +514,7 @@ class TestHa(unittest.TestCase):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
         self.assertEquals(self.ha.run_cycle(), 'PAUSE: DCS is not accessible')
 
-    @patch('patroni.ha.sleep', Mock())
+    @patch('time.sleep', Mock())
     def test_process_sync_replication(self):
         self.ha.has_lock = true
         mock_set_sync = self.p.set_synchronous_standby = Mock()
@@ -634,7 +634,7 @@ class TestHa(unittest.TestCase):
         mock_promote.assert_called_once()
         mock_write_sync.assert_called_once_with('other', None, index=0)
 
-    @patch('patroni.utils.sleep')
+    @patch('time.sleep')
     def test_disable_sync_when_restarting(self, mock_sleep):
         self.ha.is_synchronous_mode = true
 

tests/test_patroni.py

@@ -1,4 +1,5 @@
 import etcd
+import signal
 import sys
 import time
 import unittest
@@ -8,7 +9,7 @@ from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
 from patroni.dcs.etcd import Client
 from patroni.exceptions import DCSError
-from patroni import Patroni, main as _main
+from patroni import Patroni, main as _main, patroni_main
 from six.moves import BaseHTTPServer
 from test_etcd import SleepException, etcd_read, etcd_write
 from test_postgresql import Postgresql, psycopg2_connect
@@ -53,16 +54,47 @@ class TestPatroni(unittest.TestCase):
     @patch('time.sleep', Mock(side_effect=SleepException))
     @patch.object(etcd.Client, 'delete', Mock())
     @patch.object(Client, 'machines')
-    def test_patroni_main(self, mock_machines):
+    def test_patroni_patroni_main(self, mock_machines):
         with patch('subprocess.call', Mock(return_value=1)):
             sys.argv = ['patroni.py', 'postgres0.yml']
 
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             with patch.object(Patroni, 'run', Mock(side_effect=SleepException)):
-                self.assertRaises(SleepException, _main)
+                self.assertRaises(SleepException, patroni_main)
             with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
                 with patch('patroni.ha.Ha.is_paused', Mock(return_value=True)):
-                    _main()
+                    patroni_main()
+
+    @patch('os.getpid')
+    @patch('subprocess.Popen', )
+    @patch('patroni.patroni_main', Mock())
+    def test_patroni_main(self, mock_popen, mock_getpid):
+        mock_getpid.return_value = 2
+        _main()
+
+        mock_getpid.return_value = 1
+
+        def mock_signal(signo, handler):
+            handler(signo, None)
+
+        with patch('signal.signal', mock_signal):
+            with patch('os.waitpid', Mock(side_effect=[(1, 0), (0, 0)])):
+                _main()
+            with patch('os.waitpid', Mock(side_effect=OSError)):
+                _main()
+
+        ref = {'passtochild': lambda signo, stack_frame: 0}
+
+        def mock_sighup(signo, handler):
+            if signo == signal.SIGHUP:
+                ref['passtochild'] = handler
+
+        def mock_wait():
+            ref['passtochild'](0, None)
+
+        mock_popen.return_value.wait = mock_wait
+        with patch('signal.signal', mock_sighup), patch('os.kill', Mock()):
+            self.assertIsNone(_main())
 
     @patch('patroni.config.Config.save_cache', Mock())
     @patch('patroni.config.Config.reload_local_configuration', Mock(return_value=True))

tests/test_utils.py

@@ -2,25 +2,7 @@ import unittest
 
 from mock import Mock, patch
 from patroni.exceptions import PatroniException
-from patroni.utils import reap_children, Retry, RetryFailedError, sigchld_handler, sleep
+from patroni.utils import Retry, RetryFailedError
-
-
-def time_sleep(_):
-    sigchld_handler(None, None)
-
-
-class TestUtils(unittest.TestCase):
-
-    @patch('time.sleep', Mock())
-    def test_reap_children(self):
-        self.assertIsNone(reap_children())
-        with patch('os.waitpid', Mock(return_value=(0, 0))):
-            sigchld_handler(None, None)
-            self.assertIsNone(reap_children())
-
-    @patch('time.sleep', time_sleep)
-    def test_sleep(self):
-        self.assertIsNone(sleep(0.01))
 
 
 @patch('time.sleep', Mock())