From 5facf8638ef62c25e077e2fa3bd2079a0c246902 Mon Sep 17 00:00:00 2001 From: Anthony Scalisi Date: Mon, 20 Apr 2015 11:39:15 -0700 Subject: [PATCH 01/34] support multiple recovery params --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index b6c52c19..d682bd4e 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -160,7 +160,7 @@ 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.write("%s = '%s'\n" % (name, value)) f.close() def follow_the_leader(self, leader_hash): From 4e786c15192d0cd0a48b39a7e2d4ce7e6e1fdf01 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 12:12:37 +0200 Subject: [PATCH 02/34] Format code according to pep8 --- helpers/postgresql.py | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index d682bd4e..830d1686 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -1,4 +1,7 @@ -import os, psycopg2, re, time +import os +import psycopg2 +import re +import time import logging from urlparse import urlparse @@ -18,7 +21,8 @@ class Postgresql: self.config = config self.cursor_holder = None - self.connection_string = "postgres://%s:%s@%s:%s/postgres" % (self.replication["username"], self.replication["password"], self.host, self.port) + self.connection_string = "postgres://%s:%s@%s:%s/postgres" % ( + self.replication["username"], self.replication["password"], self.host, self.port) self.conn = None @@ -74,7 +78,7 @@ class Postgresql: 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 + {"data_dir": self.data_dir, "host": leader.hostname, "port": leader.port, "username": leader.username}) == 0 def is_leader(self): return not self.query("SELECT pg_is_in_recovery();").fetchone()[0] @@ -126,7 +130,8 @@ class Postgresql: 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_lsn - pg_last_xlog_replay_location() AS bytes;" % self.xlog_position()) xlog_diff = member_cursor.fetchone()[0] logger.info([self.name, member["hostname"], xlog_diff]) if xlog_diff < 0: @@ -178,7 +183,8 @@ recovery_target_timeline = 'latest' self.restart() 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 \"%s\" WITH REPLICATION ENCRYPTED PASSWORD '%s';" % + (self.replication["username"], self.replication["password"])) def xlog_position(self): return self.query("SELECT pg_last_xlog_replay_location();").fetchone()[0] From 5fc770a3709aaa362fe8fdb00b6c61a510ddf8fb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 12:38:18 +0200 Subject: [PATCH 03/34] Got rid of unneeded calls of os.system --- helpers/postgresql.py | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 830d1686..bc6c12d8 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -68,17 +68,18 @@ class Postgresql: return False def sync_from_leader(self, leader): + pgpass = "pgpass" leader = urlparse(leader["address"]) - f = open("./pgpass", "w") + 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() - os.system("chmod 600 pgpass") + os.chmod(pgpass, 0600) - 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 + return os.system("PGPASSFILE={pgpass} pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}".format( + pgpass=pgpass, data_dir=self.data_dir, hostname=leader.hostname, port=leader.port, username=leader.username)) == 0 def is_leader(self): return not self.query("SELECT pg_is_in_recovery();").fetchone()[0] @@ -170,10 +171,13 @@ recovery_target_timeline = 'latest' 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 + pattern = 'host={hostname} port={port}'.format(hostname=leader.hostname, port=leader.port) + with open(os.path.join(self.data_dir, 'recovery.conf', 'r')) as f: + for line in f: + if pattern in line: + return + self.write_recovery_conf(leader_hash) + self.restart() def promote(self): return os.system("pg_ctl promote -w -D %s" % self.data_dir) == 0 From 6f3476a924a6ea2f123954301e25a945b6913caa Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 13:09:11 +0200 Subject: [PATCH 04/34] Try to replace % formatting with .format() --- helpers/postgresql.py | 90 ++++++++++++++++++++++--------------------- 1 file changed, 46 insertions(+), 44 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index bc6c12d8..feb75e80 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -1,15 +1,24 @@ +import logging import os import psycopg2 import re import time -import logging - -from urlparse import urlparse +import urlparse logger = logging.getLogger(__name__) +def parseurl(url): + r = urlparse.urlparse(url) + return { + 'hostname': r.hostname, + 'port': r.port, + 'username': r.username, + 'password': r.password, + } + + class Postgresql: def __init__(self, config): @@ -60,7 +69,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('initdb -D ' + self.data_dir) == 0: self.write_pg_hba() return True @@ -68,52 +77,49 @@ class Postgresql: return False def sync_from_leader(self, leader): - pgpass = "pgpass" - 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(), 0644) + f.write('{hostname}:{port}:*:{username}:{password}\n'.format(**r)) - os.chmod(pgpass, 0600) - - return os.system("PGPASSFILE={pgpass} pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}".format( - pgpass=pgpass, data_dir=self.data_dir, hostname=leader.hostname, port=leader.port, username=leader.username)) == 0 + return os.system('PGPASSFILE={pgpass} pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format( + pgpass=pgpass, data_dir=self.data_dir, **r)) == 0 def is_leader(self): 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('pg_ctl status -D {} > /dev/null'.format(self.data_dir)) == 0 def start(self): if self.is_running(): 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())) + command_code = os.system('postgres -D {} {} &'.format(self.data_dir, self.server_options())) time.sleep(5) return command_code != 0 def stop(self): - return os.system("pg_ctl stop -w -D %s -m fast -w" % self.data_dir) != 0 + return os.system('pg_ctl stop -w -m fast -D ' + self.data_dir) != 0 def reload(self): - return os.system("pg_ctl reload -w -D %s" % self.data_dir) == 0 + return os.system('pg_ctl reload -w -D ' + self.data_dir) == 0 def restart(self): - return os.system("pg_ctl restart -w -D %s -m fast" % self.data_dir) == 0 + return os.system('pg_ctl restart -w -m fast -D ' + self.data_dir) == 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 = '-c listen_addresses={} -c port={}'.format(self.host, self.port) + for setting, value in self.config['parameters'].iteritems(): + options += ' -c "{}={}"'.format(setting, value) return options def is_healthy(self): @@ -132,9 +138,9 @@ class Postgresql: 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()) + 'SELECT %s::pg_lsn - pg_last_xlog_replay_location() AS bytes', (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 @@ -149,29 +155,25 @@ 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"]}) - f.close() + with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f: + f.write('host replication {} {} md5'.format(self.replication['username'], self.replication['network'])) def write_recovery_conf(self, leader_hash): - leader = urlparse(leader_hash["address"]) + r = parseurl(leader_hash['address']) - f = open("%s/recovery.conf" % self.data_dir, "w") - f.write(""" + with open(os.path.join(self.data_dir, 'recovery.conf'), 'w') as f: + 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' +primary_slot_name = '{recovery_slot}' +primary_conninfo = 'user={username} password={password} host={hostname} port={port} sslmode=prefer sslcompression=1' 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'\n" % (name, value)) - f.close() +""".format(recovery_slot=self.name, **r)) + 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"]) - pattern = 'host={hostname} port={port}'.format(hostname=leader.hostname, port=leader.port) + r = parseurl(leader_hash['address']) + pattern = 'host={hostname} port={port}'.format(**r) with open(os.path.join(self.data_dir, 'recovery.conf', 'r')) as f: for line in f: if pattern in line: @@ -180,7 +182,7 @@ recovery_target_timeline = 'latest' self.restart() def promote(self): - return os.system("pg_ctl promote -w -D %s" % self.data_dir) == 0 + return os.system('pg_ctl promote -w -D ' + self.data_dir) == 0 def demote(self, leader): self.write_recovery_conf(leader) @@ -191,4 +193,4 @@ recovery_target_timeline = 'latest' (self.replication["username"], self.replication["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()").fetchone()[0] From e108fec30a7f6d47cb4140647ecf3ff100e83392 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 13:48:16 +0200 Subject: [PATCH 05/34] Try to replace % formatting with .format() --- helpers/postgresql.py | 36 ++++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index feb75e80..752659e0 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -13,7 +13,7 @@ def parseurl(url): r = urlparse.urlparse(url) return { 'hostname': r.hostname, - 'port': r.port, + 'port': r.port or 5432, 'username': r.username, 'password': r.password, } @@ -22,22 +22,22 @@ def parseurl(url): class Postgresql: def __init__(self, config): - self.name = config["name"] - self.host, self.port = config["listen"].split(":") - self.data_dir = config["data_dir"] - self.replication = config["replication"] + self.name = config['name'] + self.host, self.port = config['listen'].split(':') + self.data_dir = config['data_dir'] + self.replication = config['replication'] self.config = config self.cursor_holder = None - self.connection_string = "postgres://%s:%s@%s:%s/postgres" % ( - self.replication["username"], self.replication["password"], self.host, self.port) + self.connection_string = 'postgres://{username}:{password}@{listen}/postgres'.format( + listen=self.config['listen'], **self.replication) self.conn = None 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() @@ -46,8 +46,8 @@ 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): max_attempts = 0 @@ -88,14 +88,14 @@ class Postgresql: pgpass=pgpass, data_dir=self.data_dir, **r)) == 0 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 {} > /dev/null'.format(self.data_dir)) == 0 def start(self): if self.is_running(): - logger.error("Cannot start PostgreSQL because one is already running.") + logger.error('Cannot start PostgreSQL because one is already running.') return False pid_path = os.path.join(self.data_dir, 'postmaster.pid') @@ -124,17 +124,17 @@ class Postgresql: 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( @@ -156,7 +156,7 @@ class Postgresql: def write_pg_hba(self): with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f: - f.write('host replication {} {} md5'.format(self.replication['username'], self.replication['network'])) + f.write('host replication {username} {network} md5'.format(**self.replication)) def write_recovery_conf(self, leader_hash): r = parseurl(leader_hash['address']) @@ -190,7 +190,7 @@ recovery_target_timeline = 'latest' def create_replication_user(self): self.query("CREATE USER \"%s\" WITH REPLICATION ENCRYPTED PASSWORD '%s';" % - (self.replication["username"], self.replication["password"])) + (self.replication['username'], self.replication['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()').fetchone()[0] From 0c776f5d860e00bbf2a0cab2da930fc68e7ad27a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 14:11:37 +0200 Subject: [PATCH 06/34] Small bugfixes --- helpers/postgresql.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 752659e0..429b1a5d 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -23,7 +23,6 @@ class Postgresql: def __init__(self, config): self.name = config['name'] - self.host, self.port = config['listen'].split(':') self.data_dir = config['data_dir'] self.replication = config['replication'] @@ -81,7 +80,7 @@ class Postgresql: pgpass = 'pgpass' with open(pgpass, 'w') as f: - os.fchmod(f.fileno(), 0644) + os.fchmod(f.fileno(), 0600) f.write('{hostname}:{port}:*:{username}:{password}\n'.format(**r)) return os.system('PGPASSFILE={pgpass} pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format( @@ -117,7 +116,8 @@ class Postgresql: return os.system('pg_ctl restart -w -m fast -D ' + self.data_dir) == 0 def server_options(self): - options = '-c listen_addresses={} -c port={}'.format(self.host, self.port) + host, port = self.config['listen'].split(':') + options = '-c listen_addresses={} -c port={}'.format(host, port) for setting, value in self.config['parameters'].iteritems(): options += ' -c "{}={}"'.format(setting, value) return options @@ -174,7 +174,7 @@ recovery_target_timeline = 'latest' def follow_the_leader(self, leader_hash): r = parseurl(leader_hash['address']) pattern = 'host={hostname} port={port}'.format(**r) - with open(os.path.join(self.data_dir, 'recovery.conf', 'r')) as f: + with open(os.path.join(self.data_dir, 'recovery.conf'), 'r') as f: for line in f: if pattern in line: return From c75bb2b297a4dc59329576a3f27317b84f6d2aa5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 14:39:15 +0200 Subject: [PATCH 07/34] Format code according to pep8 --- helpers/ha.py | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/helpers/ha.py b/helpers/ha.py index c154a2d1..1c2426d8 100644 --- a/helpers/ha.py +++ b/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 CurrentLeaderError, HealthiestMemberError +from psycopg2 import OperationalError logger = logging.getLogger(__name__) @@ -13,7 +12,9 @@ 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 @@ -31,7 +32,7 @@ class Ha: return self.etcd.am_i_leader(self.state_handler.name) def fetch_current_leader(self): - return self.etcd.current_leader() + return self.etcd.current_leader() def run_cycle(self): try: @@ -42,7 +43,6 @@ class Ha: 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(): @@ -58,7 +58,6 @@ class Ha: 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() @@ -81,11 +80,11 @@ class Ha: self.state_handler.start() return "postgresql was stopped. starting again." return "no action. not healthy enough to do anything." - except helpers.errors.CurrentLeaderError: + except CurrentLeaderError: logger.error("failed to fetch current leader from etcd") - except psycopg2.OperationalError: + except OperationalError: logger.error("Error communicating with Postgresql. Will try again.") - except helpers.errors.HealthiestMemberError: + except HealthiestMemberError: logger.error("failed to determine healthiest member fromt etcd") def run(self): From a6d5732da3e858234bd947a7e3e1e25f6d9f5110 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 14:45:49 +0200 Subject: [PATCH 08/34] Inerit HealthiestMemberError and CurrentLeaderError from EtcdError --- helpers/errors.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/helpers/errors.py b/helpers/errors.py index 1aa0e32b..dd1bc698 100644 --- a/helpers/errors.py +++ b/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 From 83ff6986d5276c487807e91b7e02899c37238046 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 14:49:07 +0200 Subject: [PATCH 09/34] Format code according to pep8 --- helpers/etcd.py | 31 ++++++++++++++++++------------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 88673c1e..c9deff1c 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -1,11 +1,16 @@ -import urllib2, json, os, time +import urllib2 +import json +import time import logging + +from helpers.errors import CurrentLeaderError from urllib import urlencode -import helpers.errors logger = logging.getLogger(__name__) + class Etcd: + def __init__(self, config): self.scope = config["scope"] self.host = config["host"] @@ -49,7 +54,7 @@ class Etcd: except urllib2.HTTPError as e: if e.code == 404: return None - raise helpers.errors.CurrentLeaderError("Etcd is not responding properly") + raise CurrentLeaderError("Etcd is not responding properly") def members(self): try: @@ -63,17 +68,17 @@ class Etcd: except urllib2.HTTPError as e: if e.code == 404: return None - raise helpers.errors.CurrentLeaderError("Etcd is not responding properly") + raise CurrentLeaderError("Etcd is not responding properly") def touch_member(self, member, connection_string): self.put_client_path("/members/%s" % 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}) is None def attempt_to_acquire_leader(self, value): try: - return self.put_client_path("/leader", {"value": value, "ttl": self.ttl, "prevExist": False}) == None + return self.put_client_path("/leader", {"value": value, "ttl": self.ttl, "prevExist": False}) is None except urllib2.HTTPError as e: if e.code == 412: logger.info("Could not take out TTL lock: %s" % e) @@ -98,15 +103,15 @@ class Etcd: 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 + # 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 def race(self, path, value): try: - return self.put_client_path(path, {"prevExist": False, "value": value}) == None + return self.put_client_path(path, {"prevExist": False, "value": value}) is None except urllib2.HTTPError: return False From 1545cc834d043c1f715c9a0e5764c40d29deee53 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 14:57:01 +0200 Subject: [PATCH 10/34] Try to replace % formatting with .format() --- helpers/etcd.py | 21 ++++++++++----------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index c9deff1c..5201dde2 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -12,9 +12,8 @@ logger = logging.getLogger(__name__) 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) def get_client_path(self, path, max_attempts=1): attempts = 0 @@ -27,7 +26,7 @@ class Etcd: 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)) + logger.info('Failed to return %s, trying again. (%s of %s)', path, attempts, max_attempts) time.sleep(3) else: raise e @@ -43,14 +42,14 @@ class Etcd: opener.open(request) def client_url(self, path): - return "http://%s/v2/keys/service/%s%s" % (self.host, self.scope, path) + return self.base_client_url + path def current_leader(self): try: - hostname = self.get_client_path("/leader")["node"]["value"] - address = self.get_client_path("/members/%s" % hostname)["node"]["value"] + hostname = self.get_client_path('/leader')['node']['value'] + address = self.get_client_path('/members/' + hostname)['node']['value'] - return {"hostname": hostname, "address": address} + return {'hostname': hostname, 'address': address} except urllib2.HTTPError as e: if e.code == 404: return None @@ -71,7 +70,7 @@ class Etcd: raise CurrentLeaderError("Etcd is not responding properly") def touch_member(self, member, connection_string): - self.put_client_path("/members/%s" % member, {"value": connection_string}) + 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}) is None @@ -81,7 +80,7 @@ class Etcd: return self.put_client_path("/leader", {"value": value, "ttl": self.ttl, "prevExist": False}) is None except urllib2.HTTPError as e: if e.code == 412: - logger.info("Could not take out TTL lock: %s" % e) + logger.info('Could not take out TTL lock: %s', e) return False def update_leader(self, value): @@ -105,7 +104,7 @@ class Etcd: 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)) + logger.info('Lock owner: %s; I am %s', reponse["node"]["value"], value) return reponse["node"]["value"] == value # except Exception as e: # return False From 1d5f3b1fff408bf50622c43b7d365e6294b9ae4b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 16:21:49 +0200 Subject: [PATCH 11/34] Do not write primary_conninfo into recovery.conf during start --- governor.py | 2 +- helpers/postgresql.py | 22 ++++++++++++---------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/governor.py b/governor.py index 57591274..b06cf799 100755 --- a/governor.py +++ b/governor.py @@ -55,7 +55,7 @@ if postgresql.data_directory_empty(): else: time.sleep(5) else: - postgresql.write_recovery_conf({"address": "postgres://169.0.0.1:5432"}) + postgresql.write_recovery_conf(None) postgresql.start() while True: diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 429b1a5d..1de58303 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -25,14 +25,15 @@ class Postgresql: self.name = config['name'] self.data_dir = config['data_dir'] self.replication = config['replication'] + self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') self.config = config - self.cursor_holder = None self.connection_string = 'postgres://{username}:{password}@{listen}/postgres'.format( listen=self.config['listen'], **self.replication) self.conn = None + self.cursor_holder = None def cursor(self): if not self.cursor_holder: @@ -159,22 +160,23 @@ class Postgresql: f.write('host replication {username} {network} md5'.format(**self.replication)) def write_recovery_conf(self, leader_hash): - r = parseurl(leader_hash['address']) - - with open(os.path.join(self.data_dir, 'recovery.conf'), 'w') as f: - f.write(""" -standby_mode = 'on' + with open(self.recovery_conf, 'w') as f: + f.write("""standby_mode = 'on' +recovery_target_timeline = 'latest' +""") + if leader_hash and 'address' in leader_hash: + r = parseurl(leader_hash['address']) + f.write(""" primary_slot_name = '{recovery_slot}' primary_conninfo = 'user={username} password={password} host={hostname} port={port} sslmode=prefer sslcompression=1' -recovery_target_timeline = 'latest' """.format(recovery_slot=self.name, **r)) - for name, value in self.config.get('recovery_conf', {}).iteritems(): - f.write("{} = '{}'\n".format(name, value)) + for name, value in self.config.get('recovery_conf', {}).iteritems(): + f.write("{} = '{}'\n".format(name, value)) def follow_the_leader(self, leader_hash): r = parseurl(leader_hash['address']) pattern = 'host={hostname} port={port}'.format(**r) - with open(os.path.join(self.data_dir, 'recovery.conf'), 'r') as f: + with open(self.recovery_conf, 'r') as f: for line in f: if pattern in line: return From 2fd16141447c57b2229aff34f0eedd284819140b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 17:52:39 +0200 Subject: [PATCH 12/34] query method now allow to execute sql with parameters --- helpers/postgresql.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 1de58303..ca59172d 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -49,11 +49,11 @@ class Postgresql: 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: @@ -191,8 +191,8 @@ primary_conninfo = 'user={username} password={password} host={hostname} port={po self.restart() 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 xlog_position(self): return self.query('SELECT pg_last_xlog_replay_location()').fetchone()[0] From 40b20ce5d73bb93003857c8869fe73d118f398d2 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 May 2015 18:26:28 +0200 Subject: [PATCH 13/34] Start postmaster and run initdb with pg_ctl It will wait utill postgres started up and helps to get rid from magic time.sleep(5) in start method. --- helpers/postgresql.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index ca59172d..f41513b3 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -26,6 +26,7 @@ class Postgresql: self.data_dir = config['data_dir'] self.replication = config['replication'] self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') + self._pg_ctl = 'pg_ctl -w -D ' + self.data_dir self.config = config @@ -69,7 +70,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 ' + self.data_dir) == 0: + if os.system(self._pg_ctl + ' initdb') == 0: self.write_pg_hba() return True @@ -91,7 +92,7 @@ class Postgresql: return not self.query('SELECT pg_is_in_recovery()').fetchone()[0] def is_running(self): - return os.system('pg_ctl status -D {} > /dev/null'.format(self.data_dir)) == 0 + return os.system(self._pg_ctl + ' status > /dev/null') == 0 def start(self): if self.is_running(): @@ -103,24 +104,23 @@ class Postgresql: os.remove(pid_path) logger.info('Removed %s', pid_path) - command_code = os.system('postgres -D {} {} &'.format(self.data_dir, self.server_options())) - time.sleep(5) - return command_code != 0 + print(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) + return os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0 def stop(self): - return os.system('pg_ctl stop -w -m fast -D ' + self.data_dir) != 0 + return os.system(self._pg_ctl + ' stop') != 0 def reload(self): - return os.system('pg_ctl reload -w -D ' + self.data_dir) == 0 + return os.system(self._pg_ctl + ' reload') == 0 def restart(self): - return os.system('pg_ctl restart -w -m fast -D ' + self.data_dir) == 0 + return os.system(self._pg_ctl + ' restart -m fast') == 0 def server_options(self): host, port = self.config['listen'].split(':') - options = '-c listen_addresses={} -c port={}'.format(host, port) + options = '--listen_addresses={} --port={}'.format(host, port) for setting, value in self.config['parameters'].iteritems(): - options += ' -c "{}={}"'.format(setting, value) + options += " --{}='{}'".format(setting, value) return options def is_healthy(self): @@ -184,7 +184,7 @@ primary_conninfo = 'user={username} password={password} host={hostname} port={po self.restart() def promote(self): - return os.system('pg_ctl promote -w -D ' + self.data_dir) == 0 + return os.system(self._pg_ctl + ' promote') == 0 def demote(self, leader): self.write_recovery_conf(leader) From 61f0cb2fb42d3e9d21cdd25e9759d6204f6f5389 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 8 May 2015 12:28:00 +0200 Subject: [PATCH 14/34] dempote and follow_the_leader have the same functionality Now methods supporting the case when leader is not defined. There would be no primary_conninfo in recovery.conf. I like it more comparing to postgres://169.0.0.1:5432 --- helpers/postgresql.py | 39 ++++++++++++++++++++++++++------------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index f41513b3..8a119933 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -104,7 +104,6 @@ class Postgresql: os.remove(pid_path) logger.info('Removed %s', pid_path) - print(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) return os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0 def stop(self): @@ -159,27 +158,42 @@ class Postgresql: with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f: f.write('host replication {username} {network} md5'.format(**self.replication)) + @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) + + def check_recovery_conf(self, leader_hash): + if not os.path.isfile(self.recovery_conf): + return False + + pattern = leader_hash and 'address' in leader_hash and self.primary_conninfo(leader_hash['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_hash): with open(self.recovery_conf, 'w') as f: f.write("""standby_mode = 'on' recovery_target_timeline = 'latest' """) if leader_hash and 'address' in leader_hash: - r = parseurl(leader_hash['address']) f.write(""" -primary_slot_name = '{recovery_slot}' -primary_conninfo = 'user={username} password={password} host={hostname} port={port} sslmode=prefer sslcompression=1' -""".format(recovery_slot=self.name, **r)) +primary_slot_name = '{}' +primary_conninfo = '{}' +""".format(self.name, self.primary_conninfo(leader_hash['address']))) for name, value in self.config.get('recovery_conf', {}).iteritems(): f.write("{} = '{}'\n".format(name, value)) def follow_the_leader(self, leader_hash): - r = parseurl(leader_hash['address']) - pattern = 'host={hostname} port={port}'.format(**r) - with open(self.recovery_conf, 'r') as f: - for line in f: - if pattern in line: - return + if self.check_recovery_conf(leader_hash): + return self.write_recovery_conf(leader_hash) self.restart() @@ -187,8 +201,7 @@ primary_conninfo = 'user={username} password={password} host={hostname} port={po return os.system(self._pg_ctl + ' promote') == 0 def demote(self, leader): - self.write_recovery_conf(leader) - self.restart() + self.follow_the_leader(leader) def create_replication_user(self): self.query('CREATE USER "{}" WITH REPLICATION ENCRYPTED PASSWORD %s'.format( From 810808a68af6c063c2d818d759d98352de6b15df Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 8 May 2015 12:44:28 +0200 Subject: [PATCH 15/34] Format code according to pep8. replace get_client_path("/members?recursive=true") with members() replace do-plpgsq block by simple sql --- governor.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/governor.py b/governor.py index b06cf799..e976798c 100755 --- a/governor.py +++ b/governor.py @@ -1,6 +1,10 @@ #!/usr/bin/env python -import sys, os, yaml, time, urllib2, atexit +import sys +import yaml +import time +import urllib2 +import atexit import logging from helpers.etcd import Etcd @@ -19,6 +23,8 @@ postgresql = Postgresql(config["postgresql"]) ha = Ha(postgresql, etcd) # stop postgresql on script exit + + def stop_postgresql(): postgresql.stop() atexit.register(stop_postgresql) @@ -63,9 +69,10 @@ while True: # 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}) + for member in etcd.members(): + if member['hostname'] != postgresql.name: + postgresql.query("""SELECT pg_create_physical_replication_slot(%s) + WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", member['hostname'], member['hostname']) time.sleep(config["loop_wait"]) From 72010f68cc633e97b108d9eb79235d8a9b1a37d8 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 8 May 2015 16:21:09 +0200 Subject: [PATCH 16/34] Bugfix: two calls has_lock and update_lock are not atomic Having a lock a few moments ago, doesn't mean that you will be able to update it. --- helpers/ha.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/helpers/ha.py b/helpers/ha.py index 1c2426d8..ed9f11f4 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -59,9 +59,7 @@ class Ha: 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() - + if self.has_lock() and self.update_lock() if not self.state_handler.is_leader(): self.state_handler.promote() return "promoted self to leader because i had the session lock" From 3a6afe5a9bec2fa041a9c255de92db49217be380 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 8 May 2015 16:32:57 +0200 Subject: [PATCH 17/34] Set environment variable PGPASSFILE via os.environ before running pg_basebackup and unset it afterwards --- helpers/postgresql.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 8a119933..c21d6440 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -85,8 +85,12 @@ class Postgresql: os.fchmod(f.fileno(), 0600) f.write('{hostname}:{port}:*:{username}:{password}\n'.format(**r)) - return os.system('PGPASSFILE={pgpass} pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format( - pgpass=pgpass, data_dir=self.data_dir, **r)) == 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] From 588c951a861549aa09575e9433d41989abbfa68b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 11 May 2015 08:48:51 +0200 Subject: [PATCH 18/34] Return True after success call of update_leader and check return code --- helpers/etcd.py | 1 + helpers/ha.py | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 5201dde2..e3603b0c 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -86,6 +86,7 @@ class Etcd: def update_leader(self, value): try: self.put_client_path("/leader", {"value": value, "ttl": self.ttl, "prevValue": value}) + return True except urllib2.HTTPError: logger.error("Error updating TTL on ETCD for primary.") return False diff --git a/helpers/ha.py b/helpers/ha.py index ed9f11f4..725918b3 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -59,7 +59,7 @@ class Ha: 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() and self.update_lock() + if self.has_lock() and self.update_lock(): if not self.state_handler.is_leader(): self.state_handler.promote() return "promoted self to leader because i had the session lock" From 8cfbdcfcd4173b77ac8434e8c1cb984926f4fc49 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 11 May 2015 09:44:57 +0200 Subject: [PATCH 19/34] Track list of already existing physical replication slots Drop replication slot when it was removed from etcd. Execute pg_create_physical_replication_slot only when something new appeared in etcd. --- helpers/ha.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/ha.py b/helpers/ha.py index 725918b3..633ececf 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -74,7 +74,7 @@ class Ha: self.state_handler.follow_the_leader(self.fetch_current_leader()) return "no action. i am a secondary and i am following a leader" else: - if not self.state_handler.is_running(): + if not self.state_handler.is_running(): # XXX is_running == is_healthy self.state_handler.start() return "postgresql was stopped. starting again." return "no action. not healthy enough to do anything." From 0b288a7f3e54fd91946d1ef8761efb5d55154830 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 11 May 2015 09:48:15 +0200 Subject: [PATCH 20/34] Track list of already existing physical replication slots Drop replication slot when it was removed from etcd. Execute pg_create_physical_replication_slot only when something new appeared in etcd. --- governor.py | 7 ++----- helpers/postgresql.py | 24 +++++++++++++++++++++++- 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/governor.py b/governor.py index e976798c..e145531c 100755 --- a/governor.py +++ b/governor.py @@ -69,10 +69,7 @@ while True: # create replication slots if postgresql.is_leader(): - for member in etcd.members(): - if member['hostname'] != postgresql.name: - postgresql.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", member['hostname'], member['hostname']) + members = [m['hostname'] for m in etcd.members() if m['hostname'] != postgresql.name] + postgresql.create_replication_slots(members) time.sleep(config["loop_wait"]) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index c21d6440..8d4f2739 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -35,6 +35,7 @@ class Postgresql: self.conn = None self.cursor_holder = None + self.members = [] # list of already existing replication slots def cursor(self): if not self.cursor_holder: @@ -100,6 +101,7 @@ class Postgresql: def start(self): if self.is_running(): + self.load_replication_slots() logger.error('Cannot start PostgreSQL because one is already running.') return False @@ -108,7 +110,9 @@ class Postgresql: os.remove(pid_path) logger.info('Removed %s', pid_path) - return os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 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(self._pg_ctl + ' stop') != 0 @@ -213,3 +217,21 @@ primary_conninfo = '{}' def xlog_position(self): return self.query('SELECT pg_last_xlog_replay_location()').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 From 2e3eb333cbf8128c3faf800f2339a7fa7f6b308c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 11 May 2015 09:48:59 +0200 Subject: [PATCH 21/34] Shutdown fast --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 8d4f2739..d8653e96 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -115,7 +115,7 @@ class Postgresql: return ret def stop(self): - return os.system(self._pg_ctl + ' stop') != 0 + return os.system(self._pg_ctl + ' stop -m fast') != 0 def reload(self): return os.system(self._pg_ctl + ' reload') == 0 From 6d17d39776d76ff602f09fdd9be967b9dac672a4 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 11 May 2015 10:06:18 +0200 Subject: [PATCH 22/34] Backport from compose/governor: compare xlog positions based on bytes since 0/000000 based on feedback --- helpers/postgresql.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index d8653e96..dd8ad4c6 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -146,7 +146,7 @@ class Postgresql: 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(), )) + "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]) if xlog_diff < 0: @@ -216,7 +216,7 @@ primary_conninfo = '{}' self.replication['username']), self.replication['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'") From eb1721f65ff51b803910705a677eade1d425628c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 11 May 2015 15:15:08 +0200 Subject: [PATCH 23/34] Minimize amount of requests to etcd --- governor.py | 5 -- helpers/etcd.py | 163 +++++++++++++++++++++--------------------- helpers/ha.py | 55 ++++++++------ helpers/postgresql.py | 20 +++--- 4 files changed, 128 insertions(+), 115 deletions(-) diff --git a/governor.py b/governor.py index e145531c..bbca37a0 100755 --- a/governor.py +++ b/governor.py @@ -67,9 +67,4 @@ else: while True: logging.info(ha.run_cycle()) - # create replication slots - if postgresql.is_leader(): - members = [m['hostname'] for m in etcd.members() if m['hostname'] != postgresql.name] - postgresql.create_replication_slots(members) - time.sleep(config["loop_wait"]) diff --git a/helpers/etcd.py b/helpers/etcd.py index e3603b0c..7e8182f8 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -1,117 +1,120 @@ -import urllib2 -import json -import time import logging +import requests +import time -from helpers.errors import CurrentLeaderError -from urllib import urlencode +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.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 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/' + hostname)['node']['value'] - - return {'hostname': hostname, 'address': address} - except urllib2.HTTPError as e: - if e.code == 404: - return None - raise 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: + cluster = self.get_cluster() + if not cluster['leader'] or not cluster['leader'].address: return None + return cluster['leader'] + except: raise CurrentLeaderError("Etcd is not responding properly") def touch_member(self, member, connection_string): - self.put_client_path('/members/' + member, {"value": connection_string}) + 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}) is 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}) is 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}) - return True - 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}) is None - except urllib2.HTTPError: - return False + return self.put_client_path(path, value=value, prevExist=False) diff --git a/helpers/ha.py b/helpers/ha.py index 633ececf..da8e0d18 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -2,7 +2,7 @@ import inspect import logging import time -from helpers.errors import CurrentLeaderError, HealthiestMemberError +from helpers.errors import EtcdError, HealthiestMemberError from psycopg2 import OperationalError logger = logging.getLogger(__name__) @@ -18,6 +18,10 @@ 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) @@ -26,60 +30,71 @@ 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) + logger.info('Lock owner: %s; I am %s', self.cluster.leader.hostname, self.state_handler.name) + return self.cluster.leader.hostname == 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(): + self.load_cluster_from_etcd() if self.is_unlocked(): - if self.state_handler.is_healthiest_node(self.etcd.members()): + 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 by acquiring session lock" return "acquired session lock as a leader" else: + self.load_cluster_from_etcd() if self.state_handler.is_leader(): - self.state_handler.demote(self.fetch_current_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()) + 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.state_handler.demote(self.fetch_current_leader()) + self.demote() return "demoting self because i am not the healthiest node" else: - self.state_handler.follow_the_leader(self.fetch_current_leader()) + self.follow_the_leader() return "following a different leader because i am not the healthiest node" else: if self.has_lock() and self.update_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" + 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.state_handler.demote(self.fetch_current_leader()) + self.demote() return "demoting self because i do not have the lock and i was a leader" else: - self.state_handler.follow_the_leader(self.fetch_current_leader()) + self.follow_the_leader() return "no action. i am a secondary and i am following a leader" else: - if not self.state_handler.is_running(): # XXX is_running == is_healthy + if not self.state_handler.is_running(): # XXX is_running == is_healthy self.state_handler.start() return "postgresql was stopped. starting again." return "no action. not healthy enough to do anything." - except CurrentLeaderError: - logger.error("failed to fetch current leader from etcd") + except EtcdError: + logger.error("Error communicating with Etcd") except OperationalError: logger.error("Error communicating with Postgresql. Will try again.") except HealthiestMemberError: diff --git a/helpers/postgresql.py b/helpers/postgresql.py index dd8ad4c6..a4a368af 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -139,10 +139,10 @@ class Postgresql: 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( @@ -171,11 +171,11 @@ class Postgresql: r = parseurl(leader_url) return 'user={username} password={password} host={hostname} port={port} sslmode=prefer sslcompression=1'.format(**r) - def check_recovery_conf(self, leader_hash): + def check_recovery_conf(self, leader): if not os.path.isfile(self.recovery_conf): return False - pattern = leader_hash and 'address' in leader_hash and self.primary_conninfo(leader_hash['address']) + pattern = leader and leader.address and self.primary_conninfo(leader.address) with open(self.recovery_conf, 'r') as f: for line in f: @@ -186,23 +186,23 @@ class Postgresql: return not pattern - def write_recovery_conf(self, leader_hash): + def write_recovery_conf(self, leader): with open(self.recovery_conf, 'w') as f: f.write("""standby_mode = 'on' recovery_target_timeline = 'latest' """) - if leader_hash and 'address' in leader_hash: + if leader and leader.address: f.write(""" primary_slot_name = '{}' primary_conninfo = '{}' -""".format(self.name, self.primary_conninfo(leader_hash['address']))) +""".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): - if self.check_recovery_conf(leader_hash): + def follow_the_leader(self, leader): + if self.check_recovery_conf(leader): return - self.write_recovery_conf(leader_hash) + self.write_recovery_conf(leader) self.restart() def promote(self): From e2faf641d53896e77503fe445b5887fe4ae8e738 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 11 May 2015 16:50:21 +0200 Subject: [PATCH 24/34] Start slave with the correct recovery.conf on governor start Also governor is able to pick up already running master and slave instances without restarting them. In case is you have a lock and postgres is not running behaviour remains the same: it will start master in read only mode and then promote if it still has the lock. --- governor.py | 22 ++++--------- helpers/etcd.py | 2 +- helpers/ha.py | 83 ++++++++++++++++++++++++++----------------------- 3 files changed, 51 insertions(+), 56 deletions(-) diff --git a/governor.py b/governor.py index bbca37a0..b3b2061f 100755 --- a/governor.py +++ b/governor.py @@ -1,11 +1,10 @@ #!/usr/bin/env python -import sys -import yaml -import time -import urllib2 import atexit import logging +import sys +import time +import yaml from helpers.etcd import Etcd from helpers.postgresql import Postgresql @@ -30,14 +29,9 @@ def stop_postgresql(): 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) +while not etcd.touch_member(postgresql.name, postgresql.connection_string): + logging.info("waiting on etcd") + time.sleep(5) # is data directory empty? if postgresql.data_directory_empty(): @@ -60,11 +54,7 @@ if postgresql.data_directory_empty(): synced_from_leader = True else: time.sleep(5) -else: - postgresql.write_recovery_conf(None) - postgresql.start() while True: logging.info(ha.run_cycle()) - time.sleep(config["loop_wait"]) diff --git a/helpers/etcd.py b/helpers/etcd.py index 7e8182f8..112e8e4d 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -103,7 +103,7 @@ class Etcd: raise CurrentLeaderError("Etcd is not responding properly") def touch_member(self, member, connection_string): - self.put_client_path('/members/' + 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) diff --git a/helpers/ha.py b/helpers/ha.py index da8e0d18..88b9bd6c 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -44,55 +44,60 @@ class Ha: def run_cycle(self): try: - if self.state_handler.is_healthy(): - 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 by acquiring session lock" - return "acquired session lock as a leader" - else: - self.load_cluster_from_etcd() - if self.state_handler.is_leader(): - self.demote() - return "demoted self due after trying and failing to obtain lock" - else: - self.follow_the_leader() - return "following new leader after trying and failing to obtain lock" + self.load_cluster_from_etcd() + if self.is_unlocked(): + if not self.state_handler.is_healthy(): + return 'no action. not healthy enough to do anything.' + elif 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 by acquiring session lock" + return "acquired session lock as a leader" else: self.load_cluster_from_etcd() if self.state_handler.is_leader(): self.demote() - return "demoting self because i am not the healthiest node" + return "demoted self due after trying and failing to obtain lock" else: self.follow_the_leader() - return "following a different leader because i am not the healthiest node" + return "following new leader after trying and failing to obtain lock" else: - 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]) + self.load_cluster_from_etcd() + if self.state_handler.is_leader(): + self.demote() + return "demoting self because i am not the healthiest node" 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" + self.follow_the_leader() + return "following a different leader because i am not the healthiest node" else: - if not self.state_handler.is_running(): # XXX is_running == is_healthy + if self.has_lock() and not self.state_handler.is_healthy(): + self.state_handler.write_recovery_conf(None) self.state_handler.start() - return "postgresql was stopped. starting again." - return "no action. not healthy enough to do anything." + self.load_cluster_from_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 not self.state_handler.is_healthy(): + self.state_handler.write_recovery_conf(self.cluster.leader) + self.state_handler.start() + return 'starting as a secondary' + elif 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: From be7677a1b8dcaacbb600b7f758f4318c518b7b27 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 12 May 2015 09:08:45 +0200 Subject: [PATCH 25/34] Bugfix, hostname and address are not keys of dict but properties of Member object --- helpers/etcd.py | 4 ++-- helpers/postgresql.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 112e8e4d..181246d4 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -96,9 +96,9 @@ class Etcd: def current_leader(self): try: cluster = self.get_cluster() - if not cluster['leader'] or not cluster['leader'].address: + if not cluster.leader or not cluster.leader.address: return None - return cluster['leader'] + return cluster.leader except: raise CurrentLeaderError("Etcd is not responding properly") diff --git a/helpers/postgresql.py b/helpers/postgresql.py index a4a368af..59c0f002 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -79,7 +79,7 @@ class Postgresql: return False def sync_from_leader(self, leader): - r = parseurl(leader['address']) + r = parseurl(leader.address) pgpass = 'pgpass' with open(pgpass, 'w') as f: @@ -148,7 +148,7 @@ class Postgresql: 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 From 9dcbc75fd26d03b4a8d599bfba3e8b9892d87959 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 12 May 2015 10:19:45 +0200 Subject: [PATCH 26/34] Bugfix. It wasn't possible to start the old cluster if all its members were shutdown. Basically this is rollback to the original decision tree with the small exception: 1 - If leader is defined and it's not me - then slave would be started immidiately with the correct recovery conf. 2 - If leader is defined in and it's my host, then it will start instance in readonly (but without primary_conninfo in recovery.conf) 3 - And the third case - if the leader is not defined - it also will start instance in readonly, without primary_conninfo. After performing 2 or 3 it will perform usual decision tree. --- helpers/ha.py | 52 +++++++++++++++++++++++++-------------------------- 1 file changed, 25 insertions(+), 27 deletions(-) diff --git a/helpers/ha.py b/helpers/ha.py index 88b9bd6c..0399b0fe 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -45,65 +45,63 @@ class Ha: def run_cycle(self): try: 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 not self.state_handler.is_healthy(): - return 'no action. not healthy enough to do anything.' - elif self.state_handler.is_healthiest_node(self.cluster.members): + 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 by acquiring session lock" - return "acquired session lock as a leader" + return 'promoted self to leader by acquiring session lock' + return 'acquired session lock as a leader' else: self.load_cluster_from_etcd() if self.state_handler.is_leader(): self.demote() - return "demoted self due after trying and failing to obtain lock" + return 'demoted self due after trying and failing to obtain lock' else: self.follow_the_leader() - return "following new leader after trying and failing to obtain lock" + 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" + 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" + return 'following a different leader because i am not the healthiest node' else: - if self.has_lock() and not self.state_handler.is_healthy(): - self.state_handler.write_recovery_conf(None) - self.state_handler.start() - self.load_cluster_from_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" + return 'promoted self to leader because i had the session lock' else: - return "no action. i am the leader with the lock" + 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 not self.state_handler.is_healthy(): - self.state_handler.write_recovery_conf(self.cluster.leader) - self.state_handler.start() - return 'starting as a secondary' - elif self.state_handler.is_leader(): + 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" + 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" + return 'no action. i am a secondary and i am following a leader' except EtcdError: - logger.error("Error communicating with Etcd") + logger.error('Error communicating with Etcd') except OperationalError: - logger.error("Error communicating with Postgresql. Will try again.") + logger.error('Error communicating with Postgresql. Will try again') except HealthiestMemberError: - logger.error("failed to determine healthiest member fromt etcd") + logger.error('failed to determine healthiest member fromt etcd') def run(self): while True: From 281b0b8455c83dfec5a97fdf5ed2d71575d45111 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 12 May 2015 11:07:52 +0200 Subject: [PATCH 27/34] Bugfix in has_lock: sometimes leader can be undefined --- helpers/ha.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/helpers/ha.py b/helpers/ha.py index 0399b0fe..3131a235 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -33,8 +33,9 @@ class Ha: return not (self.cluster.leader and self.cluster.leader.hostname) def has_lock(self): - logger.info('Lock owner: %s; I am %s', self.cluster.leader.hostname, self.state_handler.name) - return self.cluster.leader.hostname == 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 demote(self): return self.state_handler.demote(self.cluster.leader) From a56b346295131c4ddd5f26bb8e2f930875de4a8a Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 12 May 2015 11:58:47 +0200 Subject: [PATCH 28/34] Hardened the health checks and implemented a real status page. For the postgresql helper, hardened the code to get "the" cursor of the postgresql instance. For the statuspage, a small status json is returned. To find out what status a PostgreSQL cluster is in we use the cursor (instead of the provided query() function), as the query function does some retrying etc. For the healthcheck we want to simple provide an answer to a simple query, if we have to reconnect, we are not *that* healthy anyway. Dropped catching exceptions in the do_GET block, as the HTTPServer will do that nicely for us anyway. --- helpers/postgresql.py | 17 +++++++---- helpers/statuspage.py | 66 +++++++++++++++++++++++++++++++------------ 2 files changed, 60 insertions(+), 23 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 4bde365c..7f4ccaad 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -12,6 +12,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.get('superuser') @@ -20,15 +27,15 @@ class Postgresql: 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://%s:%s@%s:%s/postgres" % (self.replication["username"], self.replication["password"], self.libpq_parameters['host'], self.port) self.conn = None def cursor(self): - if not self.cursor_holder: - self.conn = psycopg2.connect("postgres://%s:%s/postgres" % (self.host, self.port)) - self.conn.autocommit = True + if (self.cursor_holder is None) or self.cursor_holder.closed: + if (self.conn is None) or self.conn.closed: + self.conn = psycopg2.connect(**self.libpq_parameters) + self.conn.autocommit = True self.cursor_holder = self.conn.cursor() return self.cursor_holder diff --git a/helpers/statuspage.py b/helpers/statuspage.py index 67fdb4ab..9aae769b 100644 --- a/helpers/statuspage.py +++ b/helpers/statuspage.py @@ -2,28 +2,59 @@ # -*- 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()) - else: - self.send_response(404) - except Exception, e: - self.send_response(500) - self.end_headers() - self.wfile.write(repr(e)) + if self.path == '/pg_master': + self.pg_master() + elif self.path == '/pg_slave': + self.pg_slave() + elif self.path == '/pg_status': + self.pg_status() + else: + self.send_response(404) + + def pg_master(self): + if not self.pg_is_in_recovery(): + self.send_response(200) + return + + self.send_response(503) + + def pg_slave(self): + if self.pg_is_in_recovery(): + self.send_response(200) + return + + self.send_response(503) + + 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_replayed': res[1], + 'delay': res[2]}, 'server': {'hostaddr': res[3], 'port': res[4], 'start_time': res[5]}} + + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(json.dumps(status)) def getHTTPServer(postgresql, http_port=8081, listen_address='0.0.0.0'): @@ -36,7 +67,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') From 8b38dbd2b55733a6f6bc7fc897538b7987d8c115 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 12 May 2015 13:48:06 +0200 Subject: [PATCH 29/34] Put governor code into class and implement sigterm processing handler --- governor.py | 99 ++++++++++++++++++++++++++++++----------------------- 1 file changed, 57 insertions(+), 42 deletions(-) diff --git a/governor.py b/governor.py index b3b2061f..cafc79b3 100755 --- a/governor.py +++ b/governor.py @@ -1,7 +1,8 @@ #!/usr/bin/env python -import atexit import logging +import os +import signal import sys import time import yaml @@ -11,50 +12,64 @@ from helpers.postgresql import Postgresql from helpers.ha import Ha -logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) - -f = open(sys.argv[1], "r") -config = yaml.load(f.read()) -f.close() - -etcd = Etcd(config["etcd"]) -postgresql = Postgresql(config["postgresql"]) -ha = Ha(postgresql, etcd) - -# stop postgresql on script exit +def sigterm_handler(signo, stack_frame): + sys.exit() -def stop_postgresql(): - postgresql.stop() -atexit.register(stop_postgresql) +class Governor: -# wait for etcd to be available -while not etcd.touch_member(postgresql.name, postgresql.connection_string): - logging.info("waiting on etcd") - time.sleep(5) + def __init__(self, config): + self.nap_time = config['loop_wait'] + self.etcd = Etcd(config['etcd']) + self.postgresql = Postgresql(config['postgresql']) + self.ha = Ha(self.postgresql, self.etcd) -# 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() - 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 + 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) + + # 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) + 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()) - time.sleep(config["loop_wait"]) + def run(self): + while True: + logging.info(self.ha.run_cycle()) + time.sleep(self.nap_time) + + +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) + try: + governor.initialize() + 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() From 6043bf8c754375e7b4fc92dd5c3f70f34f03edcd Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 12 May 2015 14:07:10 +0200 Subject: [PATCH 30/34] Fix imports --- governor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/governor.py b/governor.py index 8a3d55c0..1431cc66 100755 --- a/governor.py +++ b/governor.py @@ -2,6 +2,7 @@ import logging import os +import requests import signal import sys import threading @@ -11,7 +12,7 @@ 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 def sigterm_handler(signo, stack_frame): @@ -39,7 +40,6 @@ class Governor: self.postgresql = Postgresql(config['postgresql'], aws_host_address) self.ha = Ha(self.postgresql, self.etcd) - def initialize(self): # wait for etcd to be available while not self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string): From d25fdd41a6a4f376a3c54877ec18dcd3b5790b8a Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 12 May 2015 14:14:20 +0200 Subject: [PATCH 31/34] Refactoring of the status page for the healthcheck. Less functions, as the code is readable enough without them. Always return some content to the client, instead of a response only. --- helpers/statuspage.py | 44 +++++++++++++++++++++---------------------- 1 file changed, 21 insertions(+), 23 deletions(-) diff --git a/helpers/statuspage.py b/helpers/statuspage.py index 9aae769b..ee535b54 100644 --- a/helpers/statuspage.py +++ b/helpers/statuspage.py @@ -9,27 +9,23 @@ class StatusPage(BaseHTTPRequestHandler): def do_GET(self): if self.path == '/pg_master': - self.pg_master() + if not self.pg_is_in_recovery(): + response, content = 200, 'I am currently a master' + else: + response, content = 503, 'I am not a master' elif self.path == '/pg_slave': - self.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': - self.pg_status() + response, content = 200, self.pg_status() else: - self.send_response(404) + response, content = 404, 'Page not found' - def pg_master(self): - if not self.pg_is_in_recovery(): - self.send_response(200) - return - - self.send_response(503) - - def pg_slave(self): - if self.pg_is_in_recovery(): - self.send_response(200) - return - - self.send_response(503) + self.send_response(response) + self.end_headers() + self.wfile.write(content) def pg_is_in_recovery(self): cursor = self.server.postgresql.cursor() @@ -48,13 +44,12 @@ class StatusPage(BaseHTTPRequestHandler): 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_replayed': res[1], + status = {'role': ('master' if not res[0] else 'slave'), 'recovery': {'last_transaction_timestamp': res[1], 'delay': res[2]}, 'server': {'hostaddr': res[3], 'port': res[4], 'start_time': res[5]}} - self.send_response(200) self.send_header('Content-Type', 'application/json') - self.end_headers() - self.wfile.write(json.dumps(status)) + + return json.dumps(status) def getHTTPServer(postgresql, http_port=8081, listen_address='0.0.0.0'): @@ -84,5 +79,8 @@ if __name__ == '__main__': 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() From 362b6b4fa2f23329cff0b0cb8c4f8aab09f8c2c2 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 12 May 2015 14:48:40 +0200 Subject: [PATCH 32/34] Made healtcheck port configurable, fixed standalone status page. --- governor.py | 2 +- helpers/statuspage.py | 8 +++++--- postgres0.yml | 1 + postgres1.yml | 1 + 4 files changed, 8 insertions(+), 4 deletions(-) diff --git a/governor.py b/governor.py index 1431cc66..638305df 100755 --- a/governor.py +++ b/governor.py @@ -80,7 +80,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', 8080), listen_address='0.0.0.0') http_thread = threading.Thread(target=http_server.serve_forever, args=()) http_thread.daemon = True diff --git a/helpers/statuspage.py b/helpers/statuspage.py index ee535b54..cec68432 100644 --- a/helpers/statuspage.py +++ b/helpers/statuspage.py @@ -44,8 +44,8 @@ class StatusPage(BaseHTTPRequestHandler): 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], - 'delay': res[2]}, 'server': {'hostaddr': res[3], 'port': res[4], 'start_time': res[5]}} + 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]}} self.send_header('Content-Type', 'application/json') @@ -71,8 +71,10 @@ 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: diff --git a/postgres0.yml b/postgres0.yml index e7d05c5a..14c22892 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -1,5 +1,6 @@ loop_wait: 10 aws_use_host_address: "on" +healthcheck_port: 8080 etcd: scope: batman ttl: 30 diff --git a/postgres1.yml b/postgres1.yml index f18ebd17..bb203c6b 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -1,5 +1,6 @@ loop_wait: 10 aws_use_host_address: "on" +healthcheck_port: 8081 etcd: scope: batman ttl: 30 From c4168bfeb5bef26382f4f4f56807c48f0cd8240a Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 12 May 2015 14:56:02 +0200 Subject: [PATCH 33/34] Bugfix: Headers sent before http status for the pg_status page. --- helpers/statuspage.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/helpers/statuspage.py b/helpers/statuspage.py index cec68432..2819beeb 100644 --- a/helpers/statuspage.py +++ b/helpers/statuspage.py @@ -8,6 +8,7 @@ import json class StatusPage(BaseHTTPRequestHandler): def do_GET(self): + content_type='text/plain' if self.path == '/pg_master': if not self.pg_is_in_recovery(): response, content = 200, 'I am currently a master' @@ -20,10 +21,12 @@ class StatusPage(BaseHTTPRequestHandler): 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) @@ -47,8 +50,6 @@ class StatusPage(BaseHTTPRequestHandler): 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]}} - self.send_header('Content-Type', 'application/json') - return json.dumps(status) From 2b56379b85ffea736c18087060818b69ff8086ee Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 12 May 2015 15:48:30 +0200 Subject: [PATCH 34/34] Change default port for the health check to 8008. --- governor.py | 2 +- postgres0.yml | 2 +- postgres1.yml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/governor.py b/governor.py index 638305df..2f9132cf 100755 --- a/governor.py +++ b/governor.py @@ -80,7 +80,7 @@ def main(): governor = Governor(config) # Start the http_server to serve a simple healthcheck - http_server = getHTTPServer(governor.postgresql, http_port=config.get('healtcheck_port', 8080), 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 diff --git a/postgres0.yml b/postgres0.yml index 14c22892..ca577a0b 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -1,6 +1,6 @@ loop_wait: 10 aws_use_host_address: "on" -healthcheck_port: 8080 +healthcheck_port: 8008 etcd: scope: batman ttl: 30 diff --git a/postgres1.yml b/postgres1.yml index bb203c6b..8f0e493e 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -1,6 +1,6 @@ loop_wait: 10 aws_use_host_address: "on" -healthcheck_port: 8081 +healthcheck_port: 8009 etcd: scope: batman ttl: 30