Merge branch 'master' of github.com:zalando/governor into features/refactoring

governor.py

@@ -19,6 +19,17 @@ 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
+
+
 class Governor:
 
     INSTANCE_METADATA_URL = "http://169.254.169.254/latest/meta-data/"
@@ -80,7 +91,7 @@ def main():
     governor = Governor(config)
 
     # Start the http_server to serve a simple healthcheck
-    http_server = getHTTPServer(governor.postgresql, http_port=8008, listen_address='0.0.0.0')
+    http_server = getHTTPServer(governor.postgresql, http_port=config.get('healtcheck_port', 8008), listen_address='0.0.0.0')
     http_thread = threading.Thread(target=http_server.serve_forever, args=())
     http_thread.daemon = True
 
@@ -96,4 +107,5 @@ def main():
 if __name__ == '__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)
     main()

helpers/postgresql.py

@@ -31,6 +31,13 @@ class Postgresql:
     def __init__(self, config, aws_host_address=None):
         self.name = config['name']
         self.host, self.port = config['listen'].split(':')
+        self.libpq_parameters = {
+            'host' : aws_host_address or self.host,
+            'port' : self.port,
+            'fallback_application_name' : 'Governor',
+            'connect_timeout' : 5,
+            'options' : '-c statement_timeout=2000'
+        }
         self.data_dir = config['data_dir']
         self.replication = config['replication']
         self.superuser = config['superuser']
@@ -40,9 +47,8 @@ class Postgresql:
 
         self.config = config
 
-        connection_host = aws_host_address or self.host
         self.connection_string = 'postgres://{username}:{password}@{host}:{port}/postgres'.format(
-            host=connection_host, port=self.port, **self.replication)
+            host=self.libpq_parameters['host'], port=self.port, **self.replication)
 
         self.conn = None
         self.cursor_holder = None

helpers/statuspage.py

@@ -2,28 +2,55 @@
 # -*- coding: utf-8 -*-
 
 from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
+import json
 
 
 class StatusPage(BaseHTTPRequestHandler):
 
     def do_GET(self):
-        try:
-            if self.path == '/pg_master':
-                response = (200 if self.server.postgresql.is_leader else 503)
-                self.send_response(response)
-            elif self.path == '/pg_slave':
-                response = (503 if self.server.postgresql.is_leader else 200)
-                self.send_response(response)
-            elif self.path == '/pg_status':
-                self.send_response(200)
-                self.end_headers()
-                self.wfile.write(self.server.postgresql.status())
+        content_type='text/plain'
+        if self.path == '/pg_master':
+            if not self.pg_is_in_recovery():
+                response, content = 200, 'I am currently a master'
             else:
-                self.send_response(404)
-        except Exception, e:
-            self.send_response(500)
-            self.end_headers()
-            self.wfile.write(repr(e))
+                response, content = 503, 'I am not a master'
+        elif self.path == '/pg_slave':
+            if self.pg_is_in_recovery():
+                response, content = 200, 'I am currently a slave'
+            else:
+                response, content = 503, 'I am not a slave'
+        elif self.path == '/pg_status':
+            response, content = 200, self.pg_status()
+            content_type = 'application/json'
+        else:
+            response, content = 404, 'Page not found'
+
+        self.send_response(response)
+        self.send_header('Content-Type', content_type)
+        self.end_headers()
+        self.wfile.write(content)
+
+    def pg_is_in_recovery(self):
+        cursor = self.server.postgresql.cursor()
+        cursor.execute('SELECT pg_is_in_recovery()')
+        res = cursor.fetchone()
+        return res[0]
+
+    def pg_status(self):
+        cursor = self.server.postgresql.cursor()
+        cursor.execute("""
+            SELECT pg_is_in_recovery(),
+                   to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),
+                   extract(epoch from now() - pg_last_xact_replay_timestamp()),
+                   inet_server_addr(),
+                   inet_server_port(),
+                   to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ')
+                    """)
+        res = cursor.fetchone()
+        status = {'role': ('master' if not res[0] else 'slave'), 'recovery': {'last_transaction_timestamp': res[1]},
+                  'server': {'hostaddr': res[3], 'port': res[4], 'start_time': res[5]}}
+
+        return json.dumps(status)
 
 
 def getHTTPServer(postgresql, http_port=8081, listen_address='0.0.0.0'):
@@ -36,7 +63,6 @@ def getHTTPServer(postgresql, http_port=8081, listen_address='0.0.0.0'):
 if __name__ == '__main__':
     import sys
     import logging
-    from BaseHTTPServer import HTTPServer
 
     logging.basicConfig(format='%(levelname)-6s %(asctime)s - %(message)s', level=logging.DEBUG)
     logging.debug('Starting as a standalone application')
@@ -46,13 +72,18 @@ if __name__ == '__main__':
     postgres_config = {
         'name': 'dummy',
         'listen': 'localhost:5432',
-        'data_dir': None,
+        'data_dir': 'nonsense',
         'replication': {'username': None, 'password': None},
+        'superuser': None,
+        'admin': None,
     }
     aws_host_address = None
     if len(sys.argv) > 1:
         postgres_config['listen'] = sys.argv[1]
     postgresql = Postgresql(postgres_config, aws_host_address)
 
-    getHTTPServer(postgresql, 8081, '0.0.0.0').serve_forever()
-    logging.debug('Abc')
+    http_port = 8081
+    if len(sys.argv) > 2:
+        http_port = int(sys.argv[2])
+
+    getHTTPServer(postgresql, http_port, '0.0.0.0').serve_forever()

postgres0.yml

@@ -1,5 +1,6 @@
 loop_wait: 10
 aws_use_host_address: "on"
+healthcheck_port: 8008
 etcd:
   scope: batman
   ttl: 30

postgres1.yml

@@ -1,5 +1,6 @@
 loop_wait: 10
 aws_use_host_address: "on"
+healthcheck_port: 8009
 etcd:
   scope: batman
   ttl: 30