Merge branch 'sleep-sigchld' of github.com:CyberDem0n/governor into features/refactoring

Conflicts:
	tests/test_governor.py

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

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:

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) == []

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)

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

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)