From aa486194ae540dea06b8db6051ed33c68ecf2b40 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 28 May 2015 15:47:51 +0200 Subject: [PATCH] Wrap time.sleep into wrapper function In case if it was interrupted by SIGCHLD but scheduled awake time is not reached it will continue sleeping. For all other signals behaviuor is not changed. --- governor.py | 40 ++++++++++++++----------------- helpers/etcd.py | 4 ++-- helpers/postgresql.py | 8 +++---- helpers/utils.py | 40 +++++++++++++++++++++++++++++++ tests/test_governor.py | 53 ++++++++++++++++++------------------------ tests/test_utils.py | 44 +++++++++++++++++++++++++++++++++++ 6 files changed, 130 insertions(+), 59 deletions(-) create mode 100644 helpers/utils.py create mode 100644 tests/test_utils.py diff --git a/governor.py b/governor.py index 85cd28e0..55cb4931 100755 --- a/governor.py +++ b/governor.py @@ -1,8 +1,6 @@ #!/usr/bin/env python - import logging import os -import signal import sys import time import yaml @@ -11,21 +9,7 @@ from helpers.api import RestApiServer from helpers.etcd import Etcd from helpers.postgresql import Postgresql from helpers.ha import Ha - - -def sigterm_handler(signo, stack_frame): - sys.exit() - - -# handle SIGCHILD, since we are the equivalent of the INIT process -def sigchld_handler(signo, stack_frame): - try: - while True: - ret = os.waitpid(-1, os.WNOHANG) - if ret == (0, 0): - break - except OSError: - pass +from helpers.utils import setup_signal_handlers, sleep class Governor: @@ -37,6 +21,7 @@ class Governor: self.ha = Ha(self.postgresql, self.etcd) host, port = config['restapi']['listen'].split(':') self.api = RestApiServer(self, config['restapi']) + self.next_run = time.time() def touch_member(self, ttl=None): connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string @@ -46,7 +31,7 @@ class Governor: # wait for etcd to be available while not self.touch_member(): logging.info('waiting on etcd') - time.sleep(5) + sleep(5) # is data directory empty? if self.postgresql.data_directory_empty(): @@ -63,22 +48,33 @@ class Governor: self.postgresql.write_recovery_conf(leader) self.postgresql.start() break - time.sleep(5) + sleep(5) elif self.postgresql.is_running(): self.postgresql.load_replication_slots() + def schedule_next_run(self): + self.next_run += self.nap_time + current_time = time.time() + nap_time = self.next_run - current_time + if nap_time <= 0: + self.next_run = current_time + else: + sleep(nap_time) + def run(self): self.api.start() + self.next_run = time.time() + while True: self.touch_member() logging.info(self.ha.run_cycle()) - time.sleep(self.nap_time) + + self.schedule_next_run() def main(): logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) - signal.signal(signal.SIGTERM, sigterm_handler) - signal.signal(signal.SIGCHLD, sigchld_handler) + setup_signal_handlers() if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]): print('Usage: {} config.yml'.format(sys.argv[0])) diff --git a/helpers/etcd.py b/helpers/etcd.py index 6da6a3bc..48c02d85 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -1,10 +1,10 @@ import logging import requests -import time from requests.exceptions import RequestException from collections import namedtuple from helpers.errors import CurrentLeaderError, EtcdError +from helpers.utils import sleep logger = logging.getLogger(__name__) @@ -42,7 +42,7 @@ class Etcd: attempts += 1 if attempts < max_attempts: logger.info('Failed to return %s, trying again. (%s of %s)', path, attempts, max_attempts) - time.sleep(3) + sleep(3) elif ex: raise ex else: diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 868e9bbf..48df8345 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -3,16 +3,14 @@ import os import psycopg2 import subprocess import sys -import time -is_py3 = sys.hexversion >= 0x03000000 +from helpers.utils import sleep -if is_py3: +if sys.hexversion >= 0x03000000: from urllib.parse import urlparse else: from urlparse import urlparse - logger = logging.getLogger(__name__) @@ -93,7 +91,7 @@ class Postgresql: max_attempts += 1 if max_attempts >= 3: raise ex - time.sleep(5) + sleep(5) def data_directory_empty(self): return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == [] diff --git a/helpers/utils.py b/helpers/utils.py new file mode 100644 index 00000000..6b26444f --- /dev/null +++ b/helpers/utils.py @@ -0,0 +1,40 @@ +import os +import signal +import sys +import time + +received_sigchld = False + + +def sigterm_handler(signo, stack_frame): + sys.exit() + + +def sigchld_handler(signo, stack_frame): + global received_sigchld + received_sigchld = True + try: + while True: + ret = os.waitpid(-1, os.WNOHANG) + if ret == (0, 0): + break + except OSError: + pass + + +def sleep(interval): + global received_sigchld + current_time = time.time() + end_time = current_time + interval + while current_time < end_time: + received_sigchld = False + time.sleep(end_time - current_time) + if not received_sigchld: # we will ignore only sigchld + break + current_time = time.time() + received_sigchld = False + + +def setup_signal_handlers(): + signal.signal(signal.SIGTERM, sigterm_handler) + signal.signal(signal.SIGCHLD, sigchld_handler) diff --git a/tests/test_governor.py b/tests/test_governor.py index 941d6d45..17db3421 100644 --- a/tests/test_governor.py +++ b/tests/test_governor.py @@ -1,4 +1,3 @@ -import os import psycopg2 import requests import subprocess @@ -7,7 +6,7 @@ import time import unittest import yaml -from governor import Governor, main, sigchld_handler, sigterm_handler +from governor import Governor, main from test_ha import true, false from test_postgresql import Postgresql, subprocess_call, psycopg2_connect from test_etcd import requests_get, requests_put, requests_delete @@ -22,10 +21,6 @@ def nop(*args, **kwargs): pass -def os_waitpid(a, b): - return (0, 0) - - def time_sleep(_): raise Exception() @@ -44,20 +39,22 @@ class TestGovernor(unittest.TestCase): requests.get = requests_get requests.put = requests_put requests.delete = requests_delete + self.time_sleep = time.sleep time.sleep = nop self.write_pg_hba = Postgresql.write_pg_hba self.write_recovery_conf = Postgresql.write_recovery_conf Postgresql.write_pg_hba = nop Postgresql.write_recovery_conf = nop BaseHTTPServer.HTTPServer.__init__ = nop + with open('postgres0.yml', 'r') as f: + config = yaml.load(f) + self.g = Governor(config) def tear_down(self): + time.sleep = self.time_sleep Postgresql.write_pg_hba = self.write_pg_hba Postgresql.write_recovery_conf = self.write_recovery_conf - def test_sigterm_handler(self): - self.assertRaises(SystemExit, sigterm_handler, None, None) - def test_governor_main(self): main() sys.argv = ['governor.py', 'postgres0.yml'] @@ -71,25 +68,21 @@ class TestGovernor(unittest.TestCase): return True def test_governor_initialize(self): - with open('postgres0.yml', 'r') as f: - config = yaml.load(f) - g = Governor(config) - g.etcd.base_client_url = 'http://remote' - g.etcd.client_url - g.postgresql.data_directory_empty = true - g.etcd.race = true - g.initialize() - g.etcd.race = false - g.initialize() - g.postgresql.data_directory_empty = false - g.touch_member = self.touch_member - g.initialize() - g.postgresql.data_directory_empty = true - time.sleep = time_sleep - g.postgresql.sync_from_leader = false - self.assertRaises(Exception, g.initialize) + self.g.etcd.base_client_url = 'http://remote' + self.g.etcd.client_url + self.g.postgresql.data_directory_empty = true + self.g.etcd.race = true + self.g.initialize() + self.g.etcd.race = false + self.g.initialize() + self.g.postgresql.data_directory_empty = false + self.g.touch_member = self.touch_member + self.g.initialize() + self.g.postgresql.data_directory_empty = true + time.sleep = time_sleep + self.g.postgresql.sync_from_leader = false + self.assertRaises(Exception, self.g.initialize) - def test_sigchld_handler(self): - sigchld_handler(None, None) - os.waitpid = os_waitpid - sigchld_handler(None, None) + def test_schedule_next_run(self): + self.g.next_run = time.time() - self.g.nap_time - 1 + self.g.schedule_next_run() diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 00000000..76dbff66 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,44 @@ +import os +import time +import unittest + +from helpers.utils import sigchld_handler, sigterm_handler, sleep + + +def nop(*args, **kwargs): + pass + + +def os_waitpid(a, b): + return (0, 0) + + +def time_sleep(_): + sigchld_handler(None, None) + + +class TestUtils(unittest.TestCase): + + def __init__(self, method_name='runTest'): + self.setUp = self.set_up + self.tearDown = self.tear_down + super(TestUtils, self).__init__(method_name) + + def set_up(self): + self.time_sleep = time.sleep + time.sleep = nop + + def tear_down(self): + time.sleep = self.time_sleep + + def test_sigterm_handler(self): + self.assertRaises(SystemExit, sigterm_handler, None, None) + + def test_sigchld_handler(self): + sigchld_handler(None, None) + os.waitpid = os_waitpid + sigchld_handler(None, None) + + def test_sleep(self): + time.sleep = time_sleep + sleep(0.01)