From 5facf8638ef62c25e077e2fa3bd2079a0c246902 Mon Sep 17 00:00:00 2001 From: Anthony Scalisi Date: Mon, 20 Apr 2015 11:39:15 -0700 Subject: [PATCH 01/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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"])