From 28b00dea16bc5cde98c3b9edb27a30906b16b230 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 22 Nov 2016 16:22:47 +0100 Subject: [PATCH] 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 | 60 ++++++++++++++++++++++++++++++---------- patroni/dcs/consul.py | 4 +-- patroni/dcs/etcd.py | 4 +-- patroni/dcs/exhibitor.py | 3 +- patroni/ha.py | 15 +++++----- patroni/utils.py | 39 +++----------------------- tests/test_ha.py | 4 +-- tests/test_patroni.py | 40 ++++++++++++++++++++++++--- tests/test_utils.py | 20 +------------- 9 files changed, 102 insertions(+), 87 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 029ae6ff..8fea4455 100644 --- a/patroni/__init__.py +++ b/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() diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 94407b10..4108f908 100644 --- a/patroni/dcs/consul.py +++ b/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 diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index c1c7258d..14dae7e4 100644 --- a/patroni/dcs/etcd.py +++ b/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): diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py index 56105af1..216dbabc 100644 --- a/patroni/dcs/exhibitor.py +++ b/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(): diff --git a/patroni/ha.py b/patroni/ha.py index a25ee50d..78b94dcc 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1,17 +1,18 @@ +import datetime import functools import json import logging import psycopg2 import requests import sys -import datetime -from threading import RLock +import time 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', diff --git a/patroni/utils.py b/patroni/utils.py index ee9ce770..3a548abf 100644 --- a/patroni/utils.py +++ b/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): - global __interrupted_sleep, __reap_children - __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 _sleep(interval): + time.sleep(interval) 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) diff --git a/tests/test_ha.py b/tests/test_ha.py index c5b01884..eb9265ef 100644 --- a/tests/test_ha.py +++ b/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 diff --git a/tests/test_patroni.py b/tests/test_patroni.py index aa2c5581..5291cd0c 100644 --- a/tests/test_patroni.py +++ b/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)) diff --git a/tests/test_utils.py b/tests/test_utils.py index b32d6b35..c1144d74 100644 --- a/tests/test_utils.py +++ b/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 - - -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)) +from patroni.utils import Retry, RetryFailedError @patch('time.sleep', Mock())