diff --git a/governor.py b/governor.py index 9fd3d238..17f7056a 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(): @@ -64,22 +49,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 e6e9ccfd..e6ba42e5 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -4,16 +4,14 @@ import psycopg2 import shutil 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__) @@ -102,7 +100,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 f819fbac..aea90c61 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,26 +68,22 @@ 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.postgresql.should_use_s3_to_create_replica = false - 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.postgresql.should_use_s3_to_create_replica = false + 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)