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())