From 5facf8638ef62c25e077e2fa3bd2079a0c246902 Mon Sep 17 00:00:00 2001 From: Anthony Scalisi Date: Mon, 20 Apr 2015 11:39:15 -0700 Subject: [PATCH 01/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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/30] 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 35b111083c9ec8d1b7536d26691d6d8d4eabe9b8 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 11 May 2015 17:55:00 +0200 Subject: [PATCH 25/30] Change the governor health check default port to 8008 in order to avoid a conflict with etcd proxy. --- governor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/governor.py b/governor.py index 680bbb71..f992babd 100755 --- a/governor.py +++ b/governor.py @@ -33,7 +33,7 @@ postgresql = Postgresql(config["postgresql"], aws_host_address) ha = Ha(postgresql, etcd) ## Start the http_server to serve a simple healthcheck -http_server = getHTTPServer(postgresql, http_port=8080, listen_address='0.0.0.0') +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() From be7677a1b8dcaacbb600b7f758f4318c518b7b27 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 12 May 2015 09:08:45 +0200 Subject: [PATCH 26/30] 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 27/30] 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 28/30] 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 8b38dbd2b55733a6f6bc7fc897538b7987d8c115 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 12 May 2015 13:48:06 +0200 Subject: [PATCH 29/30] 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/30] 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):