Merge pull request #3 from zalando/feature/status_page

Refactoring of Governor, adding of a health check, setting default ports to sane values.

governor.py

@@ -1,92 +1,99 @@
 #!/usr/bin/env python
 
-import sys, os, yaml, time, urllib2, atexit
 import logging
+import os
+import requests
+import signal
+import sys
 import threading
+import time
+import yaml
 
 from helpers.etcd import Etcd
 from helpers.postgresql import Postgresql
 from helpers.ha import Ha
-from helpers.statuspage import StatusPage, getHTTPServer
+from helpers.statuspage import getHTTPServer
 
-INSTANCE_METADATA_URL = "http://169.254.169.254/latest/meta-data/"
 
-logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+def sigterm_handler(signo, stack_frame):
+    sys.exit()
 
-f = open(sys.argv[1], "r")
-config = yaml.load(f.read())
-f.close()
 
-if config.get('aws_use_host_address', False):
-    # get host address of the AWS host via a call to
-    # http://169.254.169.254/latest/meta-data/local-ipv4
-    try:
-        aws_host_address = urllib2.urlopen(INSTANCE_METADATA_URL+"/local-ipv4").read()
-    except (urllib2.HTTPError, urllib2.URLError) as e:
-        logging.error("Error retrieiving IPv4 address from AWS instance: {0}".format(e))
+class Governor:
+
+    INSTANCE_METADATA_URL = "http://169.254.169.254/latest/meta-data/"
+
+    def __init__(self, config):
+        self.nap_time = config['loop_wait']
+        self.etcd = Etcd(config['etcd'])
         aws_host_address = None
-else:
-    aws_host_address = None
+        if config.get('aws_use_host_address', False):
+            # get host address of the AWS host via a call to
+            # http://169.254.169.254/latest/meta-data/local-ipv4
+            try:
+                response = requests.get(Governor.INSTANCE_METADATA_URL + '/local-ipv4')
+                if response.status_code == 200:
+                    aws_host_address = response.content
+            except:
+                logging.exception('Error retrieiving IPv4 address from AWS instance')
 
-etcd = Etcd(config["etcd"])
-postgresql = Postgresql(config["postgresql"], aws_host_address)
-ha = Ha(postgresql, etcd)
+        self.postgresql = Postgresql(config['postgresql'], aws_host_address)
+        self.ha = Ha(self.postgresql, self.etcd)
 
-## Start the http_server to serve a simple healthcheck
-http_server = getHTTPServer(postgresql, http_port=8008, listen_address='0.0.0.0')
-http_thread = threading.Thread(target=http_server.serve_forever, args=())
-http_thread.daemon = True
-http_thread.start()
+    def initialize(self):
+        # wait for etcd to be available
+        while not self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string):
+            logging.info('waiting on etcd')
+            time.sleep(5)
 
-# stop postgresql on script exit
-def stop_postgresql():
-    postgresql.stop()
-atexit.register(stop_postgresql)
-
-# wait for etcd to be available
-etcd_ready = False
-while not etcd_ready:
-    try:
-        etcd.touch_member(postgresql.name, postgresql.connection_string)
-        etcd_ready = True
-    except urllib2.URLError:
-        logging.info("waiting on etcd")
-        time.sleep(5)
-
-# is data directory empty?
-if postgresql.data_directory_empty():
-    # racing to initialize
-    if etcd.race("/initialize", postgresql.name):
-        postgresql.initialize()
-        etcd.take_leader(postgresql.name)
-        postgresql.start()
-        postgresql.create_replication_user()
-        postgresql.create_connection_users()
-    else:
-        synced_from_leader = False
-        while not synced_from_leader:
-            leader = etcd.current_leader()
-            if not leader:
-                time.sleep(5)
-                continue
-            if postgresql.sync_from_leader(leader):
-                postgresql.write_recovery_conf(leader)
-                postgresql.start()
-                synced_from_leader = True
+        # is data directory empty?
+        if self.postgresql.data_directory_empty():
+            # racing to initialize
+            if self.etcd.race('/initialize', self.postgresql.name):
+                self.postgresql.initialize()
+                self.etcd.take_leader(self.postgresql.name)
+                self.postgresql.start()
+                self.postgresql.create_replication_user()
             else:
-                time.sleep(5)
-else:
-    postgresql.write_recovery_conf({"address": "postgres://169.0.0.1:5432"})
-    postgresql.start()
+                while True:
+                    leader = self.etcd.current_leader()
+                    if leader and self.postgresql.sync_from_leader(leader):
+                        self.postgresql.write_recovery_conf(leader)
+                        self.postgresql.start()
+                        break
+                    time.sleep(5)
 
-while True:
-    logging.info(ha.run_cycle())
+    def run(self):
+        while True:
+            logging.info(self.ha.run_cycle())
+            time.sleep(self.nap_time)
 
-    # create replication slots
-    if postgresql.is_leader():
-        for node in etcd.get_client_path("/members?recursive=true")["node"]["nodes"]:
-            member = node["key"].split('/')[-1]
-            if member != postgresql.name:
-                postgresql.query("DO LANGUAGE plpgsql $$DECLARE somevar VARCHAR; BEGIN SELECT slot_name INTO somevar FROM pg_replication_slots WHERE slot_name = '%(slot)s' LIMIT 1; IF NOT FOUND THEN PERFORM pg_create_physical_replication_slot('%(slot)s'); END IF; END$$;" % {"slot": member})
 
-    time.sleep(config["loop_wait"])
+def main():
+    if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
+        print('Usage: {} config.yml'.format(sys.argv[0]))
+        return
+
+    with open(sys.argv[1], 'r') as f:
+        config = yaml.load(f)
+
+    governor = Governor(config)
+
+    # Start the http_server to serve a simple healthcheck
+    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
+
+    governor.initialize()
+    http_thread.start()
+
+    try:
+        governor.run()
+    finally:
+        governor.postgresql.stop()
+
+
+if __name__ == '__main__':
+    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+    signal.signal(signal.SIGTERM, sigterm_handler)
+    main()

helpers/errors.py

@@ -1,13 +1,15 @@
-class CurrentLeaderError(Exception):
+class EtcdError(Exception):
+
     def __init__(self, value):
         self.value = value
 
     def __str__(self):
         return repr(self.value)
 
-class HealthiestMemberError(Exception):
-    def __init__(self, value):
-        self.value = value
 
-    def __str__(self):
-        return repr(self.value)
+class CurrentLeaderError(EtcdError):
+    pass
+
+
+class HealthiestMemberError(EtcdError):
+    pass

helpers/etcd.py

@@ -1,112 +1,120 @@
-import urllib2, json, os, time
 import logging
-from urllib import urlencode
-import helpers.errors
+import requests
+import time
+
+from collections import namedtuple
+from helpers.errors import CurrentLeaderError, EtcdError
 
 logger = logging.getLogger(__name__)
 
+
+class Member(namedtuple('Member', 'hostname,address')):
+
+    pass
+
+
+class Cluster(namedtuple('Cluster', 'leader,members')):
+
+    pass
+
+
 class Etcd:
+
     def __init__(self, config):
-        self.scope = config["scope"]
-        self.host = config["host"]
-        self.ttl = config["ttl"]
+        self.ttl = config['ttl']
+        self.base_client_url = 'http://{host}/v2/keys/service/{scope}'.format(**config)
+        self.postgres_cluster = None
 
     def get_client_path(self, path, max_attempts=1):
         attempts = 0
         response = None
 
         while True:
+            ex = None
             try:
-                response = urllib2.urlopen(self.client_url(path)).read()
-                break
-            except (urllib2.HTTPError, urllib2.URLError) as e:
-                attempts += 1
-                if attempts < max_attempts:
-                    logger.info("Failed to return %s, trying again. (%s of %s)" % (path, attempts, max_attempts))
-                    time.sleep(3)
-                else:
-                    raise e
-        try:
-            return json.loads(response)
-        except ValueError:
-            return response
+                response = requests.get(self.client_url(path))
+                if response.status_code == 200:
+                    break
+            except Exception, e:
+                logger.exception('get_client_path')
+                ex = e
 
-    def put_client_path(self, path, data):
-        opener = urllib2.build_opener(urllib2.HTTPHandler)
-        request = urllib2.Request(self.client_url(path), data=urlencode(data).replace("false", "False"))
-        request.get_method = lambda: 'PUT'
-        opener.open(request)
+            attempts += 1
+            if attempts < max_attempts:
+                logger.info('Failed to return %s, trying again. (%s of %s)', path, attempts, max_attempts)
+                time.sleep(3)
+            elif ex:
+                raise ex
+
+        return response.json(), response.status_code
+
+    def put_client_path(self, path, **data):
+        try:
+            response = requests.put(self.client_url(path), data=data)
+            return response.status_code in [200, 201]
+        except:
+            logger.exception('PUT %s data=%s', path, data)
+            return False
 
     def client_url(self, path):
-        return "http://%s/v2/keys/service/%s%s" % (self.host, self.scope, path)
+        return self.base_client_url + path
+
+    @staticmethod
+    def find_node(node, key):
+        if not node['dir']:
+            return None
+        key = node['key'] + key
+        for n in node['nodes']:
+            if n['key'] == key:
+                return n
+        return None
+
+    def get_cluster(self):
+        try:
+            response, status_code = self.get_client_path('?recursive=true')
+            if status_code == 200:
+                leader = None
+                members = self.find_node(response['node'], '/members')
+                members = [Member(n['key'].split('/')[-1], n['value']) for n in members['nodes']] if members else []
+
+                leader_node = self.find_node(response['node'], '/leader')
+                if leader_node:
+                    for m in members:
+                        if m.hostname == leader_node['value']:
+                            leader = m
+                            break
+                    if not leader:
+                        leader = Member(leader['value'], None)
+                return Cluster(leader, members)
+            elif status_code == 404:
+                return Cluster(None, [])
+        except:
+            logger.exception('get_cluster')
+
+        raise EtcdError('Etcd is not responding properly')
 
     def current_leader(self):
         try:
-            hostname = self.get_client_path("/leader")["node"]["value"]
-            address = self.get_client_path("/members/%s" % hostname)["node"]["value"]
-
-            return {"hostname": hostname, "address": address}
-        except urllib2.HTTPError as e:
-            if e.code == 404:
+            cluster = self.get_cluster()
+            if not cluster.leader or not cluster.leader.address:
                 return None
-            raise helpers.errors.CurrentLeaderError("Etcd is not responding properly")
-
-    def members(self):
-        try:
-            members = []
-
-            r = self.get_client_path("/members?recursive=true")
-            for node in r["node"]["nodes"]:
-                members.append({"hostname": node["key"].split('/')[-1], "address": node["value"]})
-
-            return members
-        except urllib2.HTTPError as e:
-            if e.code == 404:
-                return None
-            raise helpers.errors.CurrentLeaderError("Etcd is not responding properly")
+            return cluster.leader
+        except:
+            raise CurrentLeaderError("Etcd is not responding properly")
 
     def touch_member(self, member, connection_string):
-        self.put_client_path("/members/%s" % member, {"value": connection_string})
+        return self.put_client_path('/members/' + member, value=connection_string)
 
     def take_leader(self, value):
-        return self.put_client_path("/leader", {"value": value, "ttl": self.ttl}) == None
+        return self.put_client_path('/leader', value=value, ttl=self.ttl)
 
     def attempt_to_acquire_leader(self, value):
-        try:
-            return self.put_client_path("/leader", {"value": value, "ttl": self.ttl, "prevExist": False}) == None
-        except urllib2.HTTPError as e:
-            if e.code == 412:
-                logger.info("Could not take out TTL lock: %s" % e)
-            return False
+        ret = self.put_client_path('/leader', value=value, ttl=self.ttl, prevExist=False)
+        ret or logger.info('Could not take out TTL lock')
+        return ret
 
     def update_leader(self, value):
-        try:
-            self.put_client_path("/leader", {"value": value, "ttl": self.ttl, "prevValue": value})
-        except urllib2.HTTPError:
-            logger.error("Error updating TTL on ETCD for primary.")
-            return False
-
-    def leader_unlocked(self):
-        try:
-            self.get_client_path("/leader")
-            return False
-        except urllib2.HTTPError as e:
-            if e.code == 404:
-                return True
-            return False
-        except ValueError as e:
-            return False
-
-    def am_i_leader(self, value):
-        #try:
-           reponse = self.get_client_path("/leader")
-           logger.info("Lock owner: %s; I am %s" % (reponse["node"]["value"], value))
-           return reponse["node"]["value"] == value
-        #except Exception as e:
-            #return False
+        return self.put_client_path('/leader', value=value, ttl=self.ttl, prevValue=value)
 
     def race(self, path, value):
-        try:
-            return self.put_client_path(path, {"prevExist": False, "value": value}) == None
-        except urllib2.HTTPError:
-            return False
+        return self.put_client_path(path, value=value, prevExist=False)

helpers/ha.py

@@ -1,10 +1,9 @@
-import sys, time, re, urllib2, json, psycopg2
-import logging
-from base64 import b64decode
-
-import helpers.errors
-
 import inspect
+import logging
+import time
+
+from helpers.errors import EtcdError, HealthiestMemberError
+from psycopg2 import OperationalError
 
 logger = logging.getLogger(__name__)
 
@@ -13,10 +12,16 @@ def lineno():
     """Returns the current line number in our program."""
     return inspect.currentframe().f_back.f_lineno
 
+
 class Ha:
+
     def __init__(self, state_handler, etcd):
         self.state_handler = state_handler
         self.etcd = etcd
+        self.cluster = None
+
+    def load_cluster_from_etcd(self):
+        self.cluster = self.etcd.get_cluster()
 
     def acquire_lock(self):
         return self.etcd.attempt_to_acquire_leader(self.state_handler.name)
@@ -25,68 +30,79 @@ class Ha:
         return self.etcd.update_leader(self.state_handler.name)
 
     def is_unlocked(self):
-        return self.etcd.leader_unlocked()
+        return not (self.cluster.leader and self.cluster.leader.hostname)
 
     def has_lock(self):
-        return self.etcd.am_i_leader(self.state_handler.name)
+        lock_owner = self.cluster.leader and self.cluster.leader.hostname
+        logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name)
+        return lock_owner == self.state_handler.name
 
-    def fetch_current_leader(self):
-            return self.etcd.current_leader()
+    def demote(self):
+        return self.state_handler.demote(self.cluster.leader)
+
+    def follow_the_leader(self):
+        return self.state_handler.follow_the_leader(self.cluster.leader)
 
     def run_cycle(self):
         try:
-            if self.state_handler.is_healthy():
-                if self.is_unlocked():
-                    if self.state_handler.is_healthiest_node(self.etcd.members()):
-                        if self.acquire_lock():
-                            if not self.state_handler.is_leader():
-                                self.state_handler.promote()
-                                return "promoted self to leader by acquiring session lock"
-
-                            return "acquired session lock as a leader"
-                        else:
-                            if self.state_handler.is_leader():
-                                self.state_handler.demote(self.fetch_current_leader())
-                                return "demoted self due after trying and failing to obtain lock"
-                            else:
-                                self.state_handler.follow_the_leader(self.fetch_current_leader())
-                                return "following new leader after trying and failing to obtain lock"
-                    else:
-                        if self.state_handler.is_leader():
-                            self.state_handler.demote(self.fetch_current_leader())
-                            return "demoting self because i am not the healthiest node"
-                        else:
-                            self.state_handler.follow_the_leader(self.fetch_current_leader())
-                            return "following a different leader because i am not the healthiest node"
-
-                else:
-                    if self.has_lock():
-                        self.update_lock()
+            self.load_cluster_from_etcd()
+            if not self.state_handler.is_healthy():
+                has_lock = self.has_lock()
+                self.state_handler.write_recovery_conf(None if has_lock else self.cluster.leader)
+                self.state_handler.start()
+                if not has_lock:
+                    return 'started as a secondary'
+                logging.info('started as readonly because i had the session lock')
+                self.load_cluster_from_etcd()
 
+            if self.is_unlocked():
+                if self.state_handler.is_healthiest_node(self.cluster.members):
+                    if self.acquire_lock():
                         if not self.state_handler.is_leader():
                             self.state_handler.promote()
-                            return "promoted self to leader because i had the session lock"
-                        else:
-                            return "no action.  i am the leader with the lock"
+                            return 'promoted self to leader by acquiring session lock'
+                        return 'acquired session lock as a leader'
                     else:
-                        logger.info("does not have lock")
+                        self.load_cluster_from_etcd()
                         if self.state_handler.is_leader():
-                            self.state_handler.demote(self.fetch_current_leader())
-                            return "demoting self because i do not have the lock and i was a leader"
+                            self.demote()
+                            return 'demoted self due after trying and failing to obtain lock'
                         else:
-                            self.state_handler.follow_the_leader(self.fetch_current_leader())
-                            return "no action.  i am a secondary and i am following a leader"
+                            self.follow_the_leader()
+                            return 'following new leader after trying and failing to obtain lock'
+                else:
+                    self.load_cluster_from_etcd()
+                    if self.state_handler.is_leader():
+                        self.demote()
+                        return 'demoting self because i am not the healthiest node'
+                    else:
+                        self.follow_the_leader()
+                        return 'following a different leader because i am not the healthiest node'
             else:
-                if not self.state_handler.is_running():
-                    self.state_handler.start()
-                    return "postgresql was stopped.  starting again."
-                return "no action.  not healthy enough to do anything."
-        except helpers.errors.CurrentLeaderError:
-            logger.error("failed to fetch current leader from etcd")
-        except psycopg2.OperationalError:
-            logger.error("Error communicating with Postgresql.  Will try again.")
-        except helpers.errors.HealthiestMemberError:
-            logger.error("failed to determine healthiest member fromt etcd")
+                if self.has_lock() and self.update_lock():
+                    try:
+                        if not self.state_handler.is_leader():
+                            self.state_handler.promote()
+                            return 'promoted self to leader because i had the session lock'
+                        else:
+                            return 'no action.  i am the leader with the lock'
+                    finally:
+                        # create replication slots
+                        self.state_handler.create_replication_slots([m.hostname for m in self.cluster.members])
+                else:
+                    logger.info('does not have lock')
+                    if self.state_handler.is_leader():
+                        self.demote()
+                        return 'demoting self because i do not have the lock and i was a leader'
+                    else:
+                        self.follow_the_leader()
+                        return 'no action.  i am a secondary and i am following a leader'
+        except EtcdError:
+            logger.error('Error communicating with Etcd')
+        except OperationalError:
+            logger.error('Error communicating with Postgresql.  Will try again')
+        except HealthiestMemberError:
+            logger.error('failed to determine healthiest member fromt etcd')
 
     def run(self):
         while True:

helpers/postgresql.py

@@ -1,33 +1,55 @@
-import os, psycopg2, re, time
 import logging
-
-from urlparse import urlparse
+import os
+import psycopg2
+import re
+import time
+import urlparse
 
 
 logger = logging.getLogger(__name__)
 
 
+def parseurl(url):
+    r = urlparse.urlparse(url)
+    return {
+        'hostname': r.hostname,
+        'port': r.port or 5432,
+        'username': r.username,
+        'password': r.password,
+    }
+
+
 class Postgresql:
 
     def __init__(self, config, aws_host_address=None):
-        self.name = config["name"]
-        self.host, self.port = config["listen"].split(":")
-        self.data_dir = config["data_dir"]
-        self.replication = config["replication"]
-        self.superuser = config.get('superuser')
-        self.admin = config.get('admin')
+        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']
+        self.admin = config['admin']
+        self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf')
+        self._pg_ctl = 'pg_ctl -w -D ' + self.data_dir
 
         self.config = config
 
-        self.cursor_holder = None
-        connection_host = aws_host_address or self.host
-        self.connection_string = "postgres://%s:%s@%s:%s/postgres" % (self.replication["username"], self.replication["password"], connection_host, self.port)
+        self.connection_string = 'postgres://{username}:{password}@{host}:{port}/postgres'.format(
+            host=self.libpq_parameters['host'], port=self.port, **self.replication)
 
         self.conn = None
+        self.cursor_holder = None
+        self.members = []  # list of already existing replication slots
 
     def cursor(self):
         if not self.cursor_holder:
-            self.conn = psycopg2.connect("postgres://%s:%s/postgres" % (self.host, self.port))
+            self.conn = psycopg2.connect('postgres://{}/postgres'.format(self.config['listen']))
             self.conn.autocommit = True
             self.cursor_holder = self.conn.cursor()
 
@@ -36,14 +58,14 @@ class Postgresql:
     def disconnect(self):
         try:
             self.conn.close()
-        except Exception as e:
-            logger.error("Error disconnecting: %s" % e)
+        except:
+            logger.exception('Error disconnecting')
 
-    def query(self, sql):
+    def query(self, sql, *params):
         max_attempts = 0
         while True:
             try:
-                self.cursor().execute(sql)
+                self.cursor().execute(sql, params)
                 break
             except psycopg2.OperationalError as e:
                 if self.conn:
@@ -59,7 +81,7 @@ class Postgresql:
         return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []
 
     def initialize(self):
-        if os.system("initdb -D %s" % self.data_dir) == 0:
+        if os.system(self._pg_ctl + ' initdb') == 0:
             self.write_pg_hba()
 
             return True
@@ -67,71 +89,75 @@ class Postgresql:
         return False
 
     def sync_from_leader(self, leader):
-        leader = urlparse(leader["address"])
+        r = parseurl(leader.address)
 
-        f = open("./pgpass", "w")
-        f.write("%(hostname)s:%(port)s:*:%(username)s:%(password)s\n" %
-                {"hostname": leader.hostname, "port": leader.port, "username": leader.username, "password": leader.password})
-        f.close()
+        pgpass = 'pgpass'
+        with open(pgpass, 'w') as f:
+            os.fchmod(f.fileno(), 0600)
+            f.write('{hostname}:{port}:*:{username}:{password}\n'.format(**r))
 
-        os.system("chmod 600 pgpass")
-
-        return os.system("PGPASSFILE=pgpass pg_basebackup -R -D %(data_dir)s --host=%(host)s --port=%(port)s -U %(username)s" %
-                {"data_dir": self.data_dir, "host": leader.hostname, "port": leader.port, "username": leader.username}) == 0
+        try:
+            os.environ['PGPASSFILE'] = pgpass
+            return os.system('pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format(
+                data_dir=self.data_dir, **r)) == 0
+        finally:
+            os.environ.pop('PGPASSFILE')
 
     def is_leader(self):
-        return not self.query("SELECT pg_is_in_recovery();").fetchone()[0]
+        return not self.query('SELECT pg_is_in_recovery()').fetchone()[0]
 
     def is_running(self):
-        return os.system("pg_ctl status -D %s > /dev/null" % self.data_dir) == 0
+        return os.system(self._pg_ctl + ' status > /dev/null') == 0
 
     def start(self):
         if self.is_running():
-            logger.error("Cannot start PostgreSQL because one is already running.")
+            self.load_replication_slots()
+            logger.error('Cannot start PostgreSQL because one is already running.')
             return False
 
-        pid_path = "%s/postmaster.pid" % self.data_dir
+        pid_path = os.path.join(self.data_dir, 'postmaster.pid')
         if os.path.exists(pid_path):
             os.remove(pid_path)
-            logger.info("Removed %s" % pid_path)
+            logger.info('Removed %s', pid_path)
 
-        command_code = os.system("postgres -D %s %s &" % (self.data_dir, self.server_options()))
-        time.sleep(5)
-        return command_code != 0
+        ret = os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0
+        ret and self.load_replication_slots()
+        return ret
 
     def stop(self):
-        return os.system("pg_ctl stop -w -D %s -m fast -w" % self.data_dir) != 0
+        return os.system(self._pg_ctl + ' stop -m fast') != 0
 
     def reload(self):
-        return os.system("pg_ctl reload -w -D %s" % self.data_dir) == 0
+        return os.system(self._pg_ctl + ' reload') == 0
 
     def restart(self):
-        return os.system("pg_ctl restart -w -D %s -m fast" % self.data_dir) == 0
+        return os.system(self._pg_ctl + ' restart -m fast') == 0
 
     def server_options(self):
-        options = "-c listen_addresses=%s -c port=%s" % (self.host, self.port)
-        for setting, value in self.config["parameters"].iteritems():
-            options += " -c \"%s=%s\"" % (setting, value)
+        options = '--listen_addresses={} --port={}'.format(self.host, self.port)
+        for setting, value in self.config['parameters'].iteritems():
+            options += " --{}='{}'".format(setting, value)
         return options
 
     def is_healthy(self):
         if not self.is_running():
-            logger.warning("Postgresql is not running.")
+            logger.warning('Postgresql is not running.')
             return False
 
         return True
 
     def is_healthiest_node(self, members):
         for member in members:
-            if member["hostname"] == self.name:
+            if member.hostname == self.name:
                 continue
             try:
-                member_conn = psycopg2.connect(member["address"])
+                member_conn = psycopg2.connect(member.address)
                 member_conn.autocommit = True
                 member_cursor = member_conn.cursor()
-                member_cursor.execute("SELECT '%s'::pg_lsn - pg_last_xlog_replay_location() AS bytes;" % self.xlog_position())
+                member_cursor.execute(
+                    "SELECT %s - (pg_last_xlog_replay_location() - '0/0000000'::pg_lsn)", (self.xlog_position(), ))
                 xlog_diff = member_cursor.fetchone()[0]
-                logger.info([self.name, member["hostname"], xlog_diff])
+                logger.info([self.name, member.hostname, xlog_diff])
                 if xlog_diff < 0:
                     member_cursor.close()
                     return False
@@ -146,55 +172,90 @@ class Postgresql:
         return member
 
     def write_pg_hba(self):
-        f = open("%s/pg_hba.conf" % self.data_dir, "a")
-        f.write("host replication %(username)s %(network)s md5" %
-                {"username": self.replication["username"], "network": self.replication["network"]})
-        # allow TCP connections from the host's own address
-        f.write("\nhost postgres postgres samehost trust\n")
-        # allow TCP connections from the rest of the world with a password
-        f.write("\nhost all all 0.0.0.0/0 md5\n")
-        f.close()
+        with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f:
+            f.write('host replication {username} {network} md5'.format(**self.replication))
+            # allow TCP connections from the host's own address
+            f.write("\nhost postgres postgres samehost trust\n")
+            # allow TCP connections from the rest of the world with a password
+            f.write("\nhost all all 0.0.0.0/0 md5\n")
 
-    def write_recovery_conf(self, leader_hash):
-        leader = urlparse(leader_hash["address"])
+    @staticmethod
+    def primary_conninfo(leader_url):
+        r = parseurl(leader_url)
+        return 'user={username} password={password} host={hostname} port={port} sslmode=prefer sslcompression=1'.format(**r)
 
-        f = open("%s/recovery.conf" % self.data_dir, "w")
-        f.write("""
-standby_mode = 'on'
-primary_slot_name = '%(recovery_slot)s'
-primary_conninfo = 'user=%(user)s password=%(password)s host=%(hostname)s port=%(port)s sslmode=prefer sslcompression=1'
+    def check_recovery_conf(self, leader):
+        if not os.path.isfile(self.recovery_conf):
+            return False
+
+        pattern = leader and leader.address and self.primary_conninfo(leader.address)
+
+        with open(self.recovery_conf, 'r') as f:
+            for line in f:
+                if line.startswith('primary_conninfo'):
+                    if not pattern:
+                        return False
+                    return pattern in line
+
+        return not pattern
+
+    def write_recovery_conf(self, leader):
+        with open(self.recovery_conf, 'w') as f:
+            f.write("""standby_mode = 'on'
 recovery_target_timeline = 'latest'
-""" % {"recovery_slot": self.name, "user": leader.username, "password": leader.password, "hostname": leader.hostname, "port": leader.port})
-        if "recovery_conf" in self.config:
-            for name, value in self.config["recovery_conf"].iteritems():
-                f.write("%s = '%s'" % (name, value))
-        f.close()
+""")
+            if leader and leader.address:
+                f.write("""
+primary_slot_name = '{}'
+primary_conninfo = '{}'
+""".format(self.name, self.primary_conninfo(leader.address)))
+                for name, value in self.config.get('recovery_conf', {}).iteritems():
+                    f.write("{} = '{}'\n".format(name, value))
 
-    def follow_the_leader(self, leader_hash):
-        leader = urlparse(leader_hash["address"])
-        if os.system("grep 'host=%(hostname)s port=%(port)s' %(data_dir)s/recovery.conf > /dev/null" % {"hostname": leader.hostname, "port": leader.port, "data_dir": self.data_dir}) != 0:
-            self.write_recovery_conf(leader_hash)
-            self.restart()
-        return True
-
-    def promote(self):
-        return os.system("pg_ctl promote -w -D %s" % self.data_dir) == 0
-
-    def demote(self, leader):
+    def follow_the_leader(self, leader):
+        if self.check_recovery_conf(leader):
+            return
         self.write_recovery_conf(leader)
         self.restart()
 
+    def promote(self):
+        return os.system(self._pg_ctl + ' promote') == 0
+
+    def demote(self, leader):
+        self.follow_the_leader(leader)
+
     def create_replication_user(self):
-        self.query("CREATE USER \"%s\" WITH REPLICATION ENCRYPTED PASSWORD '%s';" % (self.replication["username"], self.replication["password"]))
+        self.query('CREATE USER "{}" WITH REPLICATION ENCRYPTED PASSWORD %s'.format(
+            self.replication['username']), self.replication['password'])
 
     def create_connection_users(self):
         if self.superuser:
             if 'username' in self.superuser:
-                self.query("CREATE ROLE \"{0}\" LOGIN SUPERUSER PASSWORD '{1}';".format(self.superuser["username"], self.superuser["password"]))
+                self.query("CREATE ROLE \"{0}\" LOGIN SUPERUSER PASSWORD '{1}';".format(
+                    self.superuser["username"], self.superuser["password"]))
             else:
                 self.query("ALTER ROLE postgres PASSWORD '{0}';".format(self.superuser['password']))
         if self.admin:
-            self.query("CREATE ROLE \"{0}\" LOGIN CREATEDB CREATEROLE PASSWORD '{1}';".format(self.admin["username"], self.admin["password"]))
+            self.query("CREATE ROLE \"{0}\" LOGIN CREATEDB CREATEROLE PASSWORD '{1}';".format(
+                self.admin["username"], self.admin["password"]))
 
     def xlog_position(self):
-        return self.query("SELECT pg_last_xlog_replay_location();").fetchone()[0]
+        return self.query("SELECT pg_last_xlog_replay_location() - '0/0000000'::pg_lsn").fetchone()[0]
+
+    def load_replication_slots(self):
+        cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")
+        self.members = [r[0] for r in cursor]
+
+    def create_replication_slots(self, members):
+        # drop unused slots
+        for slot in set(self.members) - set(members):
+            self.query("""SELECT pg_drop_replication_slot(%s)
+                           WHERE EXISTS(SELECT 1 FROM pg_replication_slots
+                           WHERE slot_name = %s)""", slot, slot)
+
+        # create new slots
+        for slot in set(members) - set(self.members):
+            self.query("""SELECT pg_create_physical_replication_slot(%s)
+                           WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots
+                           WHERE slot_name = %s)""", slot, slot)
+        self.members = members

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