From fa7d36da9b1984f0cbe94299c13da58ba2911cba Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 22 Oct 2015 17:21:39 -0700 Subject: [PATCH 01/25] Merged basebackup into postgresql.py; changed things to provide alternative, configurable basebackup methods. --- README.rst | 8 ++++ patroni/postgresql.py | 90 ++++++++++++++++++++++++++++++++++++++----- postgres0.yml | 2 +- postgres1.yml | 2 +- 4 files changed, 91 insertions(+), 11 deletions(-) diff --git a/README.rst b/README.rst index 7c6bc3f8..b87438a3 100644 --- a/README.rst +++ b/README.rst @@ -108,6 +108,14 @@ settings: - *recovery\_conf*: additional configuration settings written to recovery.conf when configuring follower - *parameters*: list of configuration settings for Postgres. Many of these are required for replication to work. + + - *create_replica_methods*: an ordered list of the create methods for turning a patroni node into a new replica. + "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured + as its own config item. + + - *{replica_method}* for each create_replica_method other than basebackup, you would add a configuration section + of the same name. At a minimum, this should include "command" with a full path to the actual script to be + executed. Other configuration parameters will be passed along to the script in the form "parameter=value". Replication choices ------------------- diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 81f5c02e..a68b8dfd 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -134,18 +134,64 @@ class Postgresql: @staticmethod def build_connstring(conn): - return "host={host} port={port} user={user}".format(**conn) + mconn = "" + for param, val in conn.iteritems(): + mconn = mconn + "{0}={1} ".format(param, val) + + return mconn def create_replica(self, master_connection, env): + # create the replica according to the replica_method + # defined by the user. this is a list, so we need to + # loop through all methods the user supplies connstring = self.build_connstring(master_connection) - cmd = self.config['restore'] - try: - ret = subprocess.call(shlex.split(cmd) + [self.scope, "replica", self.data_dir, connstring], env=env) - self.delete_trigger_file() - except: - logger.exception('Error when creating replica') - return 1 - return ret + env = os.environ.copy() + env['PGPASSFILE'] = 'pgpass' + # get list of replica methods from config + replica_list = self.config.get('create_replica_method', 'basebackup') + replica_methods = [rm.strip() for rm in replica_list.split(',')] + # go through them in priority order + for replica_method in replica_methods: + # if the method is basebackup, then use the built-in + if replica_method == "basebackup": + ret = self.basebackup(connstring, env) + if ret == 0: + # if basebackup succeeds, exit with success + return 0 + else: + # user-defined method; check for configuration + # not required, actually + if replica_method in self.config: + # look to see if the user has supplied a full command path + # if not, use the method name as the command + if "command" in self.config[replica_method]: + cmd = self.config[replica_method]["command"] + else: + cmd = replica_method + + # get the rest of the replica config + method_config = self.config[replica_method].copy() + # remove the command and turn it into a shlex set + del method_config["command"] + # add the default parameters + method_config.update({"scope": self.scope, + "role" : "replica", + "datadir" : self.data_dir, + "connstring" : self.connstring}) + params = ["--{0}={1}".format(arg, val) for arg, val in method_config.iteritems()] + + try: + # call script with the full set of parameters + ret = subprocess.call(shlex.split(cmd) + shlex.split(method_config), env=env) + # if we succeeded, stop + if ret == 0: + return ret + except Exception as e: + logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, e.str)) + ret = 1 + + # out of methods, return 1 + return 1 def is_leader(self): return not self.query('SELECT pg_is_in_recovery()').fetchone()[0] @@ -427,3 +473,29 @@ recovery_target_timeline = 'latest' except: logger.exception('Could not remove data directory %s', self.data_dir) self.move_data_directory() + + def basebackup(self, master_connection, env): + # creates a replica data dir using pg_basebackup. + # this is the default, built-in create_replica_method + # tries twice, then returns failure (as 1) + # uses "stream" as the xlog-method to avoid sync issues + bbfailures = 0; + maxfailures = 2; + ret = 1 + while bbfailures < maxfailures: + try: + ret = subprocess.call(['pg_basebackup', '-R', '--pgdata=%s' % self.data_dir, + '--xlog-method=stream', "--dbname=%s" % master_connection], + env=env) + if ret == 0: + break + + except Exception as e: + logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e)) + + bbfailures += 1 + if bbfailures < maxfailures: + logger.error('Trying again in 5 seconds') + time.sleep(5) + + return ret diff --git a/postgres0.yml b/postgres0.yml index f800183a..50f4065f 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -50,7 +50,7 @@ postgresql: env_dir: /home/postgres/etc/wal-e.d/env threshold_megabytes: 10240 threshold_backup_size_percentage: 30 - restore: patroni/scripts/restore.py + create_replica_method: basebackup #recovery_conf: #restore_command: cp ../wal_archive/%f %p parameters: diff --git a/postgres1.yml b/postgres1.yml index e1c3e663..dd2e2369 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -52,7 +52,7 @@ postgresql: env_dir: /home/postgres/etc/wal-e.d/env threshold_megabytes: 10240 threshold_backup_size_percentage: 30 - restore: patroni/scripts/restore.py + create_replica_method: basebackup parameters: archive_mode: "on" wal_level: hot_standby From fc68acd0ab039d777b1c5e552f33522908c66863 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Fri, 23 Oct 2015 09:37:17 -0700 Subject: [PATCH 02/25] Small changes added for testing, and failed merge from master. --- patroni/postgresql.py | 367 ++++++++++-------------------------------- 1 file changed, 87 insertions(+), 280 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 09571b80..a68b8dfd 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -9,7 +9,6 @@ import time from patroni.exceptions import PostgresConnectionException, PostgresException from patroni.utils import Retry, RetryFailedError from six.moves.urllib_parse import urlparse -from threading import Lock logger = logging.getLogger(__name__) @@ -48,8 +47,6 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] - self.pgpass = config.get('pgpass', None) or os.path.join(os.path.expanduser('~'), 'pgpass') - self.pg_rewind = config.get('pg_rewind', {}) self.callback = config.get('callbacks', {}) self.use_slots = config.get('use_slots', True) self.schedule_load_slots = self.use_slots @@ -59,6 +56,7 @@ class Postgresql: self.postmaster_pid = os.path.join(self.data_dir, 'postmaster.pid') self.trigger_file = config.get('recovery_conf', {}).get('trigger_file', None) or 'promote' self.trigger_file = os.path.abspath(os.path.join(self.data_dir, self.trigger_file)) + self._role = 'replica' self._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir] @@ -69,50 +67,8 @@ class Postgresql: self._connection = None self._cursor_holder = None - self._need_rewind = False - self._sysid = None - self.replication_slots = [] # list of already existing replication slots - self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException) - - self._state = 'stopped' - self._state_lock = Lock() - self._role = 'replica' - self._role_lock = Lock() - - if self.is_running(): - self._state = 'running' - self._role = 'master' if self.is_leader() else 'replica' - - @property - def can_rewind(self): - """ check if pg_rewind executable is there and that pg_controldata indicates - we have either wal_log_hints or checksums turned on - """ - # low-hanging fruit: check if pg_rewind configuration is there - if not self.pg_rewind or\ - not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')): - return False - - cmd = ['pg_rewind', '--help'] - try: - ret = subprocess.call(cmd, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) - if ret != 0: # pg_rewind is not there, close up the shop and go home - return False - except OSError: - return False - # check if the cluster's configuration permits pg_rewind - data = self.controldata() - return data.get('wal_log_hints setting', 'off') == 'on' or data.get('Data page checksum version', '0') != '0' - - @property - def sysid(self): - if not self._sysid: - data = self.controldata() - self._sysid = data.get('Database system identifier', "") - return self._sysid - - def require_rewind(self): - self._need_rewind = True + self.members = [] # list of already existing replication slots + self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException) def get_local_address(self): listen_addresses = self.listen_addresses.split(',') @@ -133,15 +89,9 @@ class Postgresql: def _cursor(self): if not self._cursor_holder or self._cursor_holder.closed or self._cursor_holder.connection.closed != 0: - logger.info("established a new patroni connection to the postgres cluster") self._cursor_holder = self.connection().cursor() return self._cursor_holder - def close_connection(self): - if self._cursor_holder and self._cursor_holder.connection and self._cursor_holder.connection.closed == 0: - self._cursor_holder.connection.close() - logger.info("closed patroni connection to the postgresql cluster") - def _query(self, sql, *params): cursor = None try: @@ -151,8 +101,6 @@ class Postgresql: except psycopg2.Error as e: if cursor and cursor.connection.closed == 0: raise e - if self.state == 'restarting': - raise RetryFailedError('cluster is being restarted') raise PostgresConnectionException('connection problems') def query(self, sql, *params): @@ -165,30 +113,23 @@ class Postgresql: return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == [] def initialize(self): - self.set_state('initalizing new cluster') ret = subprocess.call(self._pg_ctl + ['initdb', '-o', '--encoding=UTF8']) == 0 - if ret: - self.write_pg_hba() - else: - self.set_state('initdb failed') + ret and self.write_pg_hba() return ret def delete_trigger_file(self): os.path.exists(self.trigger_file) and os.unlink(self.trigger_file) - def write_pgpass(self, record): - with open(self.pgpass, 'w') as f: - os.fchmod(f.fileno(), 0o600) - f.write('{host}:{port}:*:{user}:{password}\n'.format(**record)) - - env = os.environ.copy() - env['PGPASSFILE'] = self.pgpass - return env - def sync_from_leader(self, leader): r = parseurl(leader.conn_url) - env = self.write_pgpass(r) + pgpass = 'pgpass' + with open(pgpass, 'w') as f: + os.fchmod(f.fileno(), 0o600) + f.write('{host}:{port}:*:{user}:{password}\n'.format(**r)) + + env = os.environ.copy() + env['PGPASSFILE'] = pgpass return self.create_replica(r, env) == 0 @staticmethod @@ -272,82 +213,40 @@ class Postgresql: @property def role(self): - with self._role_lock: - return self._role - - def set_role(self, value): - with self._role_lock: - self._role = value - - @property - def state(self): - with self._state_lock: - return self._state - - def set_state(self, value): - with self._state_lock: - self._state = value + return self._role def start(self, block_callbacks=False): if self.is_running(): + self._role = 'master' if self.is_leader() else 'replica' + self.schedule_load_slots = self.use_slots logger.error('Cannot start PostgreSQL because one is already running.') - return True + return False - self.set_role('replica' if os.path.exists(self.recovery_conf) else 'master') + self._role = 'replica' if os.path.exists(self.recovery_conf) else 'master' if os.path.exists(self.postmaster_pid): os.remove(self.postmaster_pid) logger.info('Removed %s', self.postmaster_pid) - if not block_callbacks: - self.set_state('starting') - ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0 - - self.set_state('running' if ret else 'start failed') - self.schedule_load_slots = ret and self.use_slots self.save_configuration_files() # block_callbacks is used during restart to avoid # running start/stop callbacks in addition to restart ones - ret and not block_callbacks and self.call_nowait(ACTION_ON_START) + ret and not block_callbacks and ret and self.call_nowait(ACTION_ON_START) return ret - def checkpoint(self): - try: - r = parseurl('postgres://{}/postgres'.format(self.local_address)) - r['options'] = '-c statement_timeout=0' - with psycopg2.connect(**r) as conn: - conn.autocommit = True - with conn.cursor() as cur: - cur.execute('CHECKPOINT') - except: - logging.exception('Exception during CHECKPOINT') - def stop(self, mode='fast', block_callbacks=False): - # make sure we close all connections established against - # the former node, otherwise, we might get a stalled one - # after kill -9, which would report incorrect data to - # patroni. - - self.close_connection() - if not self.is_running(): - if not block_callbacks: - self.set_state('stopped') - return True - if block_callbacks: - self.checkpoint() - else: - self.set_state('stopping') + try: + self.query('SET statement_timeout TO 0') + self.query('CHECKPOINT') + except: + logging.exception('Exception diring CHECKPOINT') ret = subprocess.call(self._pg_ctl + ['stop', '-m', mode]) == 0 # block_callbacks is used during restart to avoid # running start/stop callbacks in addition to restart ones - if not ret: - self.set_state('stop failed') - elif not block_callbacks: - self.set_state('stopped') - self.call_nowait(ACTION_ON_STOP) + ret and not block_callbacks and self.call_nowait(ACTION_ON_STOP) return ret def reload(self): @@ -356,12 +255,8 @@ class Postgresql: return ret def restart(self): - self.set_state('restarting') ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True) - if ret: - self.call_nowait(ACTION_ON_RESTART) - else: - self.set_state('restart failed ({})'.format(self.state)) + ret and self.call_nowait(ACTION_ON_RESTART) return ret def server_options(self): @@ -376,9 +271,36 @@ class Postgresql: return False return True - def check_replication_lag(self, last_leader_operation): - return (last_leader_operation if last_leader_operation else 0) - self.xlog_position() <=\ - self.config.get('maximum_lag_on_failover', 0) + def is_healthiest_node(self, cluster): + if self.is_leader(): + return True + + if cluster.last_leader_operation - self.xlog_position() > self.config.get('maximum_lag_on_failover', 0): + return False + + for member in cluster.members: + if member.name == self.name: + continue + try: + r = parseurl(member.conn_url) + member_conn = psycopg2.connect(**r) + member_conn.autocommit = True + member_cursor = member_conn.cursor() + member_cursor.execute( + "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')", + (self.xlog_position(),)) + row = member_cursor.fetchone() + member_cursor.close() + member_conn.close() + logger.error([self.name, member.name, row]) + if not row[0]: + logger.warning('Master (%s) is still alive', member.name) + return False + if row[1] < 0: + return False + except psycopg2.Error: + continue + return True def write_pg_hba(self): with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f: @@ -402,7 +324,10 @@ class Postgresql: with open(self.recovery_conf, 'r') as f: for line in f: if line.startswith('primary_conninfo'): - return pattern and (pattern in line) + if not pattern: + return False + return pattern in line + return not pattern def write_recovery_conf(self, leader): @@ -417,154 +342,40 @@ recovery_target_timeline = 'latest' for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) - def rewind(self, leader): - # prepare pg_rewind connection - r = parseurl(leader.conn_url) - r.update(self.pg_rewind) - r['user'] = r['username'] - env = self.write_pgpass(r) - pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r) - logger.info("running pg_rewind from {}".format(pc)) - pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc] - try: - ret = (subprocess.call(pg_rewind, env=env) == 0) - except: - ret = False - if ret: + def follow_the_leader(self, leader): + if not self.check_recovery_conf(leader): self.write_recovery_conf(leader) - return ret - - def controldata(self): - """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """ - result = {} - try: - data = subprocess.check_output(['pg_controldata', self.data_dir]) - if data: - data = data.decode().splitlines() - result = {l.split(':')[0].replace('Current ', '', 1): l.split(':')[1].strip() for l in data if l} - except subprocess.CalledProcessError: - logger.exception("Error when calling pg_controldata") - finally: - return result - - def read_postmaster_opts(self): - """ returns the list of option names/values from postgres.opts, Empty dict if read failed or no file """ - result = {} - try: - with open(os.path.join(self.data_dir, "postmaster.opts")) as f: - data = f.read() - opts = [opt.strip('"\n') for opt in data.split(' "')] - for opt in opts: - if '=' in opt and opt.startswith('--'): - name, val = opt.split('=', 1) - name = name.strip('-') - result[name] = val - except IOError: - logger.exception('Error when reading postmaster.opts') - finally: - return result - - def single_user_mode(self, command=None, options={}): - """ run a given command in a single-user mode. If the command is empty - then just start and stop """ - cmd = ['postgres', '--single', '-D', self.data_dir] - for opt in sorted(options): - cmd.extend(['-c', '{0}={1}'.format(opt, options[opt])]) - # need a database name to connect - cmd.append('postgres') - p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) - if p: - command and p.communicate('{}\n'.format(command)) - p.stdin.close() - return p.wait() - return 1 - - def cleanup_archive_status(self): - status_dir = os.path.join(self.data_dir, 'pg_xlog', 'archive_status') - if os.path.isdir(status_dir): - for f in os.listdir(status_dir): - path = os.path.join(status_dir, f) - try: - if os.path.islink(path): - os.unlink(path) - elif os.path.isfile(path): - os.remove(path) - except: - logger.exception("Unable to remove {}".format(path)) - - def follow_the_leader(self, leader, recovery=False): - if not self.check_recovery_conf(leader) or recovery: - change_role = (self.role == 'master') - - self._need_rewind = (self._need_rewind or change_role) and self.can_rewind - if self._need_rewind: - logger.info("set the rewind flag after demote") - self.write_recovery_conf(leader) - if not leader or not self._need_rewind: # do not rewind until the leader becomes available - ret = self.restart() - else: # we have a leader and need to rewind - if self.is_running(): - self.stop() - # at present, pg_rewind only runs when the cluster is shut down cleanly - # and not shutdown in recovery. We have to remove the recovery.conf if present - # and start/shutdown in a single user mode to emulate this. - # XXX: if recovery.conf is linked, it will be written anew as a normal file. - if os.path.islink(self.recovery_conf): - os.unlink(self.recovery_conf) - else: - os.remove(self.recovery_conf) - # Archived segments might be useful to pg_rewind, - # clean the flags that tell we should remove them. - self.cleanup_archive_status() - # Start in a single user mode and stop to produce a clean shutdown - opts = self.read_postmaster_opts() - opts['archive_mode'] = 'on' - opts['archive_command'] = 'false' - self.single_user_mode(options=opts) - if self.rewind(leader): - ret = self.start() - else: - logger.error("unable to rewind the former master") - self.remove_data_directory() - ret = True - self._need_rewind = False - change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE) - return ret - else: - return True + run_callback = self.role == 'master' + self.restart() + run_callback and self.call_nowait(ACTION_ON_ROLE_CHANGE) def save_configuration_files(self): """ - copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files - - originally stored as symlinks, those are normally skipped by pg_basebackup - - in case of WAL-E basebackup (see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239) + copy postgresql.conf to postgresql.conf.backup to preserve it in the WAL-e backup. + see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239 """ - try: - for f in self.configuration_to_save: - os.path.isfile(f) and shutil.copy(f, f + '.backup') - except: - logger.exception('unable to create backup copies of configuration files') + for f in self.configuration_to_save: + shutil.copy(f, f + '.backup') def restore_configuration_files(self): """ restore a previously saved postgresql.conf """ try: for f in self.configuration_to_save: - not os.path.isfile(f) and os.path.isfile(f+'.backup') and shutil.copy(f + '.backup', f) + shutil.copy(f + '.backup', f) except: - logger.exception('unable to restore configuration files from backup') + logger.exception('unable to restore configuration from WAL-E backup') def promote(self): if self.role == 'master': return True ret = subprocess.call(self._pg_ctl + ['promote']) == 0 if ret: - self.set_role('master') - logger.info("cleared rewind flag after becoming the leader") - self._need_rewind = False + self._role = 'master' self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret - def demote(self): - self.follow_the_leader(None) + def demote(self, leader): + self.follow_the_leader(leader) def create_replication_user(self): self.query('CREATE USER "{}" WITH REPLICATION ENCRYPTED PASSWORD %s'.format( @@ -586,34 +397,31 @@ recovery_target_timeline = 'latest' return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery() THEN pg_last_xlog_replay_location() ELSE pg_current_xlog_location() - END, '0/0')::bigint""").fetchone()[0] + END, '0/0')""").fetchone()[0] def load_replication_slots(self): if self.use_slots and self.schedule_load_slots: cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'") - self.replication_slots = [r[0] for r in cursor] + self.members = [r[0] for r in cursor] self.schedule_load_slots = False def sync_replication_slots(self, cluster): if self.use_slots: - try: - self.load_replication_slots() - slots = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else [] - # drop unused slots - for slot in set(self.replication_slots) - set(slots): - self.query("""SELECT pg_drop_replication_slot(%s) - WHERE EXISTS(SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + self.load_replication_slots() + members = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else [] + # 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(slots) - set(self.replication_slots): - self.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT 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.replication_slots = slots - except: - logger.exception('Exception when changing replication slots') + self.members = members def last_operation(self): return str(self.xlog_position()) @@ -638,7 +446,6 @@ recovery_target_timeline = 'latest' raise PostgresException("Could not bootstrap master PostgreSQL") else: if self.sync_from_leader(current_leader): - self.restore_configuration_files() self.write_recovery_conf(current_leader) ret = self.start() return ret From 364d9b5a8a711a6fbecd9424dcde9bc47ba02b9e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sat, 24 Oct 2015 13:13:23 +0200 Subject: [PATCH 03/25] Revert "Small changes added for testing, and failed merge from master." This reverts commit fc68acd0ab039d777b1c5e552f33522908c66863. --- patroni/postgresql.py | 367 ++++++++++++++++++++++++++++++++---------- 1 file changed, 280 insertions(+), 87 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index a68b8dfd..09571b80 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -9,6 +9,7 @@ import time from patroni.exceptions import PostgresConnectionException, PostgresException from patroni.utils import Retry, RetryFailedError from six.moves.urllib_parse import urlparse +from threading import Lock logger = logging.getLogger(__name__) @@ -47,6 +48,8 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] + self.pgpass = config.get('pgpass', None) or os.path.join(os.path.expanduser('~'), 'pgpass') + self.pg_rewind = config.get('pg_rewind', {}) self.callback = config.get('callbacks', {}) self.use_slots = config.get('use_slots', True) self.schedule_load_slots = self.use_slots @@ -56,7 +59,6 @@ class Postgresql: self.postmaster_pid = os.path.join(self.data_dir, 'postmaster.pid') self.trigger_file = config.get('recovery_conf', {}).get('trigger_file', None) or 'promote' self.trigger_file = os.path.abspath(os.path.join(self.data_dir, self.trigger_file)) - self._role = 'replica' self._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir] @@ -67,8 +69,50 @@ class Postgresql: self._connection = None self._cursor_holder = None - self.members = [] # list of already existing replication slots - self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException) + self._need_rewind = False + self._sysid = None + self.replication_slots = [] # list of already existing replication slots + self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException) + + self._state = 'stopped' + self._state_lock = Lock() + self._role = 'replica' + self._role_lock = Lock() + + if self.is_running(): + self._state = 'running' + self._role = 'master' if self.is_leader() else 'replica' + + @property + def can_rewind(self): + """ check if pg_rewind executable is there and that pg_controldata indicates + we have either wal_log_hints or checksums turned on + """ + # low-hanging fruit: check if pg_rewind configuration is there + if not self.pg_rewind or\ + not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')): + return False + + cmd = ['pg_rewind', '--help'] + try: + ret = subprocess.call(cmd, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) + if ret != 0: # pg_rewind is not there, close up the shop and go home + return False + except OSError: + return False + # check if the cluster's configuration permits pg_rewind + data = self.controldata() + return data.get('wal_log_hints setting', 'off') == 'on' or data.get('Data page checksum version', '0') != '0' + + @property + def sysid(self): + if not self._sysid: + data = self.controldata() + self._sysid = data.get('Database system identifier', "") + return self._sysid + + def require_rewind(self): + self._need_rewind = True def get_local_address(self): listen_addresses = self.listen_addresses.split(',') @@ -89,9 +133,15 @@ class Postgresql: def _cursor(self): if not self._cursor_holder or self._cursor_holder.closed or self._cursor_holder.connection.closed != 0: + logger.info("established a new patroni connection to the postgres cluster") self._cursor_holder = self.connection().cursor() return self._cursor_holder + def close_connection(self): + if self._cursor_holder and self._cursor_holder.connection and self._cursor_holder.connection.closed == 0: + self._cursor_holder.connection.close() + logger.info("closed patroni connection to the postgresql cluster") + def _query(self, sql, *params): cursor = None try: @@ -101,6 +151,8 @@ class Postgresql: except psycopg2.Error as e: if cursor and cursor.connection.closed == 0: raise e + if self.state == 'restarting': + raise RetryFailedError('cluster is being restarted') raise PostgresConnectionException('connection problems') def query(self, sql, *params): @@ -113,23 +165,30 @@ class Postgresql: return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == [] def initialize(self): + self.set_state('initalizing new cluster') ret = subprocess.call(self._pg_ctl + ['initdb', '-o', '--encoding=UTF8']) == 0 - ret and self.write_pg_hba() + if ret: + self.write_pg_hba() + else: + self.set_state('initdb failed') return ret def delete_trigger_file(self): os.path.exists(self.trigger_file) and os.unlink(self.trigger_file) + def write_pgpass(self, record): + with open(self.pgpass, 'w') as f: + os.fchmod(f.fileno(), 0o600) + f.write('{host}:{port}:*:{user}:{password}\n'.format(**record)) + + env = os.environ.copy() + env['PGPASSFILE'] = self.pgpass + return env + def sync_from_leader(self, leader): r = parseurl(leader.conn_url) - pgpass = 'pgpass' - with open(pgpass, 'w') as f: - os.fchmod(f.fileno(), 0o600) - f.write('{host}:{port}:*:{user}:{password}\n'.format(**r)) - - env = os.environ.copy() - env['PGPASSFILE'] = pgpass + env = self.write_pgpass(r) return self.create_replica(r, env) == 0 @staticmethod @@ -213,40 +272,82 @@ class Postgresql: @property def role(self): - return self._role + with self._role_lock: + return self._role + + def set_role(self, value): + with self._role_lock: + self._role = value + + @property + def state(self): + with self._state_lock: + return self._state + + def set_state(self, value): + with self._state_lock: + self._state = value def start(self, block_callbacks=False): if self.is_running(): - self._role = 'master' if self.is_leader() else 'replica' - self.schedule_load_slots = self.use_slots logger.error('Cannot start PostgreSQL because one is already running.') - return False + return True - self._role = 'replica' if os.path.exists(self.recovery_conf) else 'master' + self.set_role('replica' if os.path.exists(self.recovery_conf) else 'master') if os.path.exists(self.postmaster_pid): os.remove(self.postmaster_pid) logger.info('Removed %s', self.postmaster_pid) + if not block_callbacks: + self.set_state('starting') + ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0 + + self.set_state('running' if ret else 'start failed') + self.schedule_load_slots = ret and self.use_slots self.save_configuration_files() # block_callbacks is used during restart to avoid # running start/stop callbacks in addition to restart ones - ret and not block_callbacks and ret and self.call_nowait(ACTION_ON_START) + ret and not block_callbacks and self.call_nowait(ACTION_ON_START) return ret + def checkpoint(self): + try: + r = parseurl('postgres://{}/postgres'.format(self.local_address)) + r['options'] = '-c statement_timeout=0' + with psycopg2.connect(**r) as conn: + conn.autocommit = True + with conn.cursor() as cur: + cur.execute('CHECKPOINT') + except: + logging.exception('Exception during CHECKPOINT') + def stop(self, mode='fast', block_callbacks=False): + # make sure we close all connections established against + # the former node, otherwise, we might get a stalled one + # after kill -9, which would report incorrect data to + # patroni. + + self.close_connection() + if not self.is_running(): + if not block_callbacks: + self.set_state('stopped') + return True + if block_callbacks: - try: - self.query('SET statement_timeout TO 0') - self.query('CHECKPOINT') - except: - logging.exception('Exception diring CHECKPOINT') + self.checkpoint() + else: + self.set_state('stopping') ret = subprocess.call(self._pg_ctl + ['stop', '-m', mode]) == 0 # block_callbacks is used during restart to avoid # running start/stop callbacks in addition to restart ones - ret and not block_callbacks and self.call_nowait(ACTION_ON_STOP) + if not ret: + self.set_state('stop failed') + elif not block_callbacks: + self.set_state('stopped') + self.call_nowait(ACTION_ON_STOP) return ret def reload(self): @@ -255,8 +356,12 @@ class Postgresql: return ret def restart(self): + self.set_state('restarting') ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True) - ret and self.call_nowait(ACTION_ON_RESTART) + if ret: + self.call_nowait(ACTION_ON_RESTART) + else: + self.set_state('restart failed ({})'.format(self.state)) return ret def server_options(self): @@ -271,36 +376,9 @@ class Postgresql: return False return True - def is_healthiest_node(self, cluster): - if self.is_leader(): - return True - - if cluster.last_leader_operation - self.xlog_position() > self.config.get('maximum_lag_on_failover', 0): - return False - - for member in cluster.members: - if member.name == self.name: - continue - try: - r = parseurl(member.conn_url) - member_conn = psycopg2.connect(**r) - member_conn.autocommit = True - member_cursor = member_conn.cursor() - member_cursor.execute( - "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')", - (self.xlog_position(),)) - row = member_cursor.fetchone() - member_cursor.close() - member_conn.close() - logger.error([self.name, member.name, row]) - if not row[0]: - logger.warning('Master (%s) is still alive', member.name) - return False - if row[1] < 0: - return False - except psycopg2.Error: - continue - return True + def check_replication_lag(self, last_leader_operation): + return (last_leader_operation if last_leader_operation else 0) - self.xlog_position() <=\ + self.config.get('maximum_lag_on_failover', 0) def write_pg_hba(self): with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f: @@ -324,10 +402,7 @@ class Postgresql: 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 pattern and (pattern in line) return not pattern def write_recovery_conf(self, leader): @@ -342,40 +417,154 @@ recovery_target_timeline = 'latest' for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) - def follow_the_leader(self, leader): - if not self.check_recovery_conf(leader): + def rewind(self, leader): + # prepare pg_rewind connection + r = parseurl(leader.conn_url) + r.update(self.pg_rewind) + r['user'] = r['username'] + env = self.write_pgpass(r) + pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r) + logger.info("running pg_rewind from {}".format(pc)) + pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc] + try: + ret = (subprocess.call(pg_rewind, env=env) == 0) + except: + ret = False + if ret: self.write_recovery_conf(leader) - run_callback = self.role == 'master' - self.restart() - run_callback and self.call_nowait(ACTION_ON_ROLE_CHANGE) + return ret + + def controldata(self): + """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """ + result = {} + try: + data = subprocess.check_output(['pg_controldata', self.data_dir]) + if data: + data = data.decode().splitlines() + result = {l.split(':')[0].replace('Current ', '', 1): l.split(':')[1].strip() for l in data if l} + except subprocess.CalledProcessError: + logger.exception("Error when calling pg_controldata") + finally: + return result + + def read_postmaster_opts(self): + """ returns the list of option names/values from postgres.opts, Empty dict if read failed or no file """ + result = {} + try: + with open(os.path.join(self.data_dir, "postmaster.opts")) as f: + data = f.read() + opts = [opt.strip('"\n') for opt in data.split(' "')] + for opt in opts: + if '=' in opt and opt.startswith('--'): + name, val = opt.split('=', 1) + name = name.strip('-') + result[name] = val + except IOError: + logger.exception('Error when reading postmaster.opts') + finally: + return result + + def single_user_mode(self, command=None, options={}): + """ run a given command in a single-user mode. If the command is empty - then just start and stop """ + cmd = ['postgres', '--single', '-D', self.data_dir] + for opt in sorted(options): + cmd.extend(['-c', '{0}={1}'.format(opt, options[opt])]) + # need a database name to connect + cmd.append('postgres') + p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) + if p: + command and p.communicate('{}\n'.format(command)) + p.stdin.close() + return p.wait() + return 1 + + def cleanup_archive_status(self): + status_dir = os.path.join(self.data_dir, 'pg_xlog', 'archive_status') + if os.path.isdir(status_dir): + for f in os.listdir(status_dir): + path = os.path.join(status_dir, f) + try: + if os.path.islink(path): + os.unlink(path) + elif os.path.isfile(path): + os.remove(path) + except: + logger.exception("Unable to remove {}".format(path)) + + def follow_the_leader(self, leader, recovery=False): + if not self.check_recovery_conf(leader) or recovery: + change_role = (self.role == 'master') + + self._need_rewind = (self._need_rewind or change_role) and self.can_rewind + if self._need_rewind: + logger.info("set the rewind flag after demote") + self.write_recovery_conf(leader) + if not leader or not self._need_rewind: # do not rewind until the leader becomes available + ret = self.restart() + else: # we have a leader and need to rewind + if self.is_running(): + self.stop() + # at present, pg_rewind only runs when the cluster is shut down cleanly + # and not shutdown in recovery. We have to remove the recovery.conf if present + # and start/shutdown in a single user mode to emulate this. + # XXX: if recovery.conf is linked, it will be written anew as a normal file. + if os.path.islink(self.recovery_conf): + os.unlink(self.recovery_conf) + else: + os.remove(self.recovery_conf) + # Archived segments might be useful to pg_rewind, + # clean the flags that tell we should remove them. + self.cleanup_archive_status() + # Start in a single user mode and stop to produce a clean shutdown + opts = self.read_postmaster_opts() + opts['archive_mode'] = 'on' + opts['archive_command'] = 'false' + self.single_user_mode(options=opts) + if self.rewind(leader): + ret = self.start() + else: + logger.error("unable to rewind the former master") + self.remove_data_directory() + ret = True + self._need_rewind = False + change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE) + return ret + else: + return True def save_configuration_files(self): """ - copy postgresql.conf to postgresql.conf.backup to preserve it in the WAL-e backup. - see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239 + copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files + - originally stored as symlinks, those are normally skipped by pg_basebackup + - in case of WAL-E basebackup (see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239) """ - for f in self.configuration_to_save: - shutil.copy(f, f + '.backup') + try: + for f in self.configuration_to_save: + os.path.isfile(f) and shutil.copy(f, f + '.backup') + except: + logger.exception('unable to create backup copies of configuration files') def restore_configuration_files(self): """ restore a previously saved postgresql.conf """ try: for f in self.configuration_to_save: - shutil.copy(f + '.backup', f) + not os.path.isfile(f) and os.path.isfile(f+'.backup') and shutil.copy(f + '.backup', f) except: - logger.exception('unable to restore configuration from WAL-E backup') + logger.exception('unable to restore configuration files from backup') def promote(self): if self.role == 'master': return True ret = subprocess.call(self._pg_ctl + ['promote']) == 0 if ret: - self._role = 'master' + self.set_role('master') + logger.info("cleared rewind flag after becoming the leader") + self._need_rewind = False self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret - def demote(self, leader): - self.follow_the_leader(leader) + def demote(self): + self.follow_the_leader(None) def create_replication_user(self): self.query('CREATE USER "{}" WITH REPLICATION ENCRYPTED PASSWORD %s'.format( @@ -397,31 +586,34 @@ recovery_target_timeline = 'latest' return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery() THEN pg_last_xlog_replay_location() ELSE pg_current_xlog_location() - END, '0/0')""").fetchone()[0] + END, '0/0')::bigint""").fetchone()[0] def load_replication_slots(self): if self.use_slots and self.schedule_load_slots: cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'") - self.members = [r[0] for r in cursor] + self.replication_slots = [r[0] for r in cursor] self.schedule_load_slots = False def sync_replication_slots(self, cluster): if self.use_slots: - self.load_replication_slots() - members = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else [] - # 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) + try: + self.load_replication_slots() + slots = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else [] + # drop unused slots + for slot in set(self.replication_slots) - set(slots): + 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) + # create new slots + for slot in set(slots) - set(self.replication_slots): + 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 + self.replication_slots = slots + except: + logger.exception('Exception when changing replication slots') def last_operation(self): return str(self.xlog_position()) @@ -446,6 +638,7 @@ recovery_target_timeline = 'latest' raise PostgresException("Could not bootstrap master PostgreSQL") else: if self.sync_from_leader(current_leader): + self.restore_configuration_files() self.write_recovery_conf(current_leader) ret = self.start() return ret From acd21eae4c797930a79a38c62721f79401df0be7 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sun, 25 Oct 2015 18:11:45 +0100 Subject: [PATCH 04/25] Fix obvious bugs and pep8 formatting --- patroni/postgresql.py | 43 ++++++++++++++++++++----------------------- 1 file changed, 20 insertions(+), 23 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 09571b80..0b725d78 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -194,9 +194,9 @@ class Postgresql: @staticmethod def build_connstring(conn): mconn = "" - for param, val in conn.iteritems(): + for param, val in conn.items(): mconn = mconn + "{0}={1} ".format(param, val) - + return mconn def create_replica(self, master_connection, env): @@ -204,8 +204,6 @@ class Postgresql: # defined by the user. this is a list, so we need to # loop through all methods the user supplies connstring = self.build_connstring(master_connection) - env = os.environ.copy() - env['PGPASSFILE'] = 'pgpass' # get list of replica methods from config replica_list = self.config.get('create_replica_method', 'basebackup') replica_methods = [rm.strip() for rm in replica_list.split(',')] @@ -227,18 +225,18 @@ class Postgresql: cmd = self.config[replica_method]["command"] else: cmd = replica_method - + # get the rest of the replica config method_config = self.config[replica_method].copy() # remove the command and turn it into a shlex set del method_config["command"] # add the default parameters - method_config.update({"scope": self.scope, - "role" : "replica", - "datadir" : self.data_dir, - "connstring" : self.connstring}) - params = ["--{0}={1}".format(arg, val) for arg, val in method_config.iteritems()] - + method_config.update({"scope": self.scope, + "role": "replica", + "datadir": self.data_dir, + "connstring": self.connstring}) + params = ["--{0}={1}".format(arg, val) for arg, val in method_config.items()] + try: # call script with the full set of parameters ret = subprocess.call(shlex.split(cmd) + shlex.split(method_config), env=env) @@ -248,7 +246,7 @@ class Postgresql: except Exception as e: logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, e.str)) ret = 1 - + # out of methods, return 1 return 1 @@ -548,7 +546,7 @@ recovery_target_timeline = 'latest' """ restore a previously saved postgresql.conf """ try: for f in self.configuration_to_save: - not os.path.isfile(f) and os.path.isfile(f+'.backup') and shutil.copy(f + '.backup', f) + not os.path.isfile(f) and os.path.isfile(f + '.backup') and shutil.copy(f + '.backup', f) except: logger.exception('unable to restore configuration files from backup') @@ -666,29 +664,28 @@ recovery_target_timeline = 'latest' except: logger.exception('Could not remove data directory %s', self.data_dir) self.move_data_directory() - + def basebackup(self, master_connection, env): - # creates a replica data dir using pg_basebackup. + # creates a replica data dir using pg_basebackup. # this is the default, built-in create_replica_method # tries twice, then returns failure (as 1) # uses "stream" as the xlog-method to avoid sync issues - bbfailures = 0; - maxfailures = 2; + bbfailures = 0 + maxfailures = 2 ret = 1 while bbfailures < maxfailures: try: - ret = subprocess.call(['pg_basebackup', '-R', '--pgdata=%s' % self.data_dir, - '--xlog-method=stream', "--dbname=%s" % master_connection], - env=env) + ret = subprocess.call(['pg_basebackup', '-R', '--pgdata=' + self.data_dir, + '--xlog-method=stream', "--dbname=" + master_connection], env=env) if ret == 0: break - + except Exception as e: logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e)) - + bbfailures += 1 if bbfailures < maxfailures: logger.error('Trying again in 5 seconds') time.sleep(5) - + return ret From ab64ae7fe7ac804582352bb3184dd7443a1f831c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 26 Oct 2015 11:29:32 +0100 Subject: [PATCH 05/25] execute delete_trigger_file() after successfull call of create_replica() --- patroni/postgresql.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 0b725d78..bb72cb05 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -189,7 +189,9 @@ class Postgresql: r = parseurl(leader.conn_url) env = self.write_pgpass(r) - return self.create_replica(r, env) == 0 + ret = self.create_replica(r, env) == 0 + ret and self.delete_trigger_file() + return ret @staticmethod def build_connstring(conn): From 2d709a48e6fa0b331c12e42c7e887e58fa63ae51 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 26 Oct 2015 11:30:04 +0100 Subject: [PATCH 06/25] fix test_create_replica unit test --- tests/test_postgresql.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 0ed04a15..92f32a02 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -263,9 +263,13 @@ class TestPostgresql(unittest.TestCase): self.assertTrue(self.p.can_rewind) self.p.controldata = tmp + @patch('time.sleep', Mock()) def test_create_replica(self): self.p.delete_trigger_file = Mock(side_effect=OSError()) - self.assertEquals(self.p.create_replica({'host': '', 'port': '', 'user': ''}, ''), 1) + with patch('subprocess.call', Mock(side_effect=[1, 0])): + self.assertEquals(self.p.create_replica({'host': '', 'port': '', 'user': ''}, ''), 0) + with patch('subprocess.call', Mock(side_effect=[Exception(), 0])): + self.assertEquals(self.p.create_replica({'host': '', 'port': '', 'user': ''}, ''), 0) def test_create_connection_users(self): cfg = self.p.config From 06cd94b12da86ad5b1fc6d2ef86fa0f084fc7903 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Tue, 27 Oct 2015 17:35:44 -0700 Subject: [PATCH 07/25] Commit addressing several issues around this branch. Major changes to wal_e_restore script. Updated postgres0.yml to show example options. --- patroni/postgresql.py | 25 +++- patroni/scripts/wale_restore.py | 227 ++++++++++++++++++++++++++++++++ postgres0.yml | 19 ++- postgres1.yml | 23 ++-- 4 files changed, 266 insertions(+), 28 deletions(-) create mode 100644 patroni/scripts/wale_restore.py diff --git a/patroni/postgresql.py b/patroni/postgresql.py index bb72cb05..d1083c26 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -189,7 +189,7 @@ class Postgresql: r = parseurl(leader.conn_url) env = self.write_pgpass(r) - ret = self.create_replica(r, env) == 0 + ret = self.create_replica(leader, env) == 0 ret and self.delete_trigger_file() return ret @@ -201,11 +201,11 @@ class Postgresql: return mconn - def create_replica(self, master_connection, env): + def create_replica(self, leader, env): # create the replica according to the replica_method # defined by the user. this is a list, so we need to # loop through all methods the user supplies - connstring = self.build_connstring(master_connection) + connstring = leader.conn_url # get list of replica methods from config replica_list = self.config.get('create_replica_method', 'basebackup') replica_methods = [rm.strip() for rm in replica_list.split(',')] @@ -213,7 +213,7 @@ class Postgresql: for replica_method in replica_methods: # if the method is basebackup, then use the built-in if replica_method == "basebackup": - ret = self.basebackup(connstring, env) + ret = self.basebackup(leader, env) if ret == 0: # if basebackup succeeds, exit with success return 0 @@ -236,8 +236,14 @@ class Postgresql: method_config.update({"scope": self.scope, "role": "replica", "datadir": self.data_dir, - "connstring": self.connstring}) + "connstring": connstring}) params = ["--{0}={1}".format(arg, val) for arg, val in method_config.items()] + else: + cmd = replica_method + method_config = {"scope": self.scope, + "role": "replica", + "datadir": self.data_dir, + "connstring": connstring} try: # call script with the full set of parameters @@ -249,6 +255,10 @@ class Postgresql: logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, e.str)) ret = 1 + # write the recovery.conf + if ret == 0: + self.write_recovery_conf(leader) + # out of methods, return 1 return 1 @@ -667,17 +677,18 @@ recovery_target_timeline = 'latest' logger.exception('Could not remove data directory %s', self.data_dir) self.move_data_directory() - def basebackup(self, master_connection, env): + def basebackup(self, leader, env): # creates a replica data dir using pg_basebackup. # this is the default, built-in create_replica_method # tries twice, then returns failure (as 1) # uses "stream" as the xlog-method to avoid sync issues + master_connection = leader.conn_url bbfailures = 0 maxfailures = 2 ret = 1 while bbfailures < maxfailures: try: - ret = subprocess.call(['pg_basebackup', '-R', '--pgdata=' + self.data_dir, + ret = subprocess.call(['pg_basebackup', '--pgdata=' + self.data_dir, '--xlog-method=stream', "--dbname=" + master_connection], env=env) if ret == 0: break diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py new file mode 100644 index 00000000..c645f75c --- /dev/null +++ b/patroni/scripts/wale_restore.py @@ -0,0 +1,227 @@ +#!/usr/bin/python + +# sample script to clone new replicas using WAL-E restore +# falls back to pg_basebackup if WAL-E restore fails, or if +# WAL-E backup is too far behind +# note that pg_basebackup still expects to use restore from +# WAL-E for transaction logs + +# arguments are: +# - cluster scope +# - cluster role +# - master connection string +# - number of retries +# - envdir for the WALE env +# - WALE_BACKUP_THRESHOLD_MEGABYTES if WAL amount is above that - use pg_basebackup +# - WALE_BACKUP_THRESHOLD_PERCENTAGE if WAL size exceeds a certain percentage of the + +# this script depends on an envdir defining the S3 bucket (or SWIFT dir),and login +# credentials per WALE Documentation. + +# DO NOT USE with additional restore_commands; this script writes the restore command + +# latest backup size +import logging +import os +import psycopg2 +import subprocess +import sys +import argparse + + +if sys.hexversion >= 0x03000000: + long = int + +logger = logging.getLogger(__name__) + + +class Restore(object): + + def __init__(self, scope, role, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam): + self.scope = scope + self.role = role + self.master_connection = connstring + self.data_dir = datadir + self.wal_e.dir = env_dir + self.wal_e.threshold_mb = threshold_mb + self.wal_e.threshold_pct = threshold_pct + if use_iam == 1: + self.wal_e.iam_string = ' --aws-instance-profile ' + else: + self.wal_e.iam_string = '' + + def setup(self): + pass + + def replica_method(self): + return self.create_replica_with_pg_basebackup + + def replica_fallback_method(self): + return None + + def run(self): + """ creates a new replica using either pg_basebackup or WAL-E """ + method_fn = self.replica_method() + ret = method_fn() if method_fn else 1 + if ret != 0 and self.replica_fallback_method() is not None: + ret = (self.replica_fallback_method())() + return ret + + def create_replica_with_pg_basebackup(self): + try: + ret = subprocess.call(['pg_basebackup', '-R', '-D', '-x', + self.data_dir, '--host=' + self.master_connection['host'], + '--port=' + str(self.master_connection['port']), + '-U', self.master_connection['user']]) + except Exception as e: + logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e)) + return 1 + return ret + + +class WALERestore(Restore): + + def __init__(scope, role, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam): + super(WALERestore, self).__init__(scope, role, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam) + # check the environment variables + self.init_error = False + + def setup(self): + # check that we actually have an envdir + if not os.path.exists(self.wal_e.dir): + self.init_error = True + else: + self.wal_e.cmd = 'envdir {0} wal-e {1} '.\ + format(self.wal_e.dir, self.wal_e.iam_string) + + def replica_method(self): + if self.should_use_s3_to_create_replica(): + return self.create_replica_with_s3 + return None + + def replica_fallback_method(self): + return self.create_replica_with_pg_basebackup + + def should_use_s3_to_create_replica(self): + """ determine whether it makes sense to use S3 and not pg_basebackup """ + if self.init_error: + return False + + threshold_megabytes = self.wal_e.threshold_mb + threshold_backup_size_percentage = self.wal_e.threshold_pct + + try: + latest_backup = subprocess.check_output(self.wal_e.cmd.split() + ['backup-list', '--detail', 'LATEST']) + # name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start + # wal_segment_backup_stop wal_segment_offset_backup_stop + # base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z + # 20310671 00000001000000000000007F 00000040 + # 00000001000000000000007F 00000240 + backup_strings = latest_backup.splitlines() if latest_backup else () + if len(backup_strings) != 2: + return False + + names = backup_strings[0].split() + vals = backup_strings[1].split() + if (len(names) != len(vals)) or (len(names) != 7): + return False + + backup_info = dict(zip(names, vals)) + except subprocess.CalledProcessError as e: + logger.error("could not query wal-e latest backup: {}".format(e)) + return False + + try: + backup_size = backup_info['expanded_size_bytes'] + backup_start_segment = backup_info['wal_segment_backup_start'] + backup_start_offset = backup_info['wal_segment_offset_backup_start'] + except Exception as e: + logger.error("unable to get some of WALE backup parameters: {}".format(e)) + return False + + # WAL filename is XXXXXXXXYYYYYYYY000000ZZ, where X - timeline, Y - LSN logical log file, + # ZZ - 2 high digits of LSN offset. The rest of the offset is the provided decimal offset, + # that we have to convert to hex and 'prepend' to the high offset digits. + + lsn_segment = backup_start_segment[8:16] + # first 2 characters of the result are 0x and the last one is L + lsn_offset = hex((long(backup_start_segment[16:32], 16) << 24) + long(backup_start_offset))[2:-1] + + # construct the LSN from the segment and offset + backup_start_lsn = '{}/{}'.format(lsn_segment, lsn_offset) + + conn = None + cursor = None + diff_in_bytes = long(backup_size) + try: + # get the difference in bytes between the current WAL location and the backup start offset + conn = psycopg2.connect(self.master_connection) + conn.autocommit = True + cursor = conn.cursor() + cursor.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,)) + diff_in_bytes = long(cursor.fetchone()[0]) + except psycopg2.Error as e: + logger.error('could not determine difference with the master location: {}'.format(e)) + return False + finally: + cursor and cursor.close() + conn and conn.close() + + # if the size of the accumulated WAL segments is more than a certan percentage of the backup size + # or exceeds the pre-determined size - pg_basebackup is chosen instead. + return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\ + (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100) + + def write_recovery_conf_wale(self): + restore_cmd = '{0} wal_fetch "%f" "%p"'.format(self.wal_e.cmd) + with open(os.path.join(self.data_dir, 'recovery.conf'), 'w') as f: + f.write("""standby_mode = 'on' +recovery_target_timeline = 'latest' +""") + f.write("""primary_conninfo = '{}'\n""".format(self.master_connection)) + f.write("""restore_command = '{}'\n""".format(restore_cmd)) + return 0 + + def create_replica_with_s3(self): + if self.init_error: + return 1 + # if we're set up, restore the replica using fetch latest + try: + ret = subprocess.call(self.wal_e.cmd + ' backup-fetch {} LATEST'.format(self.data_dir)) + except Exception as e: + logger.error('Error when fetching backup with WAL-E: {0}'.format(e)) + return 1 + + # if success, we need to write a recovery.conf for wal-e + # this doesn't work because we need data from main + #if ret == 0: + #ret = self.write_recovery_conf_wale() + + return ret + + +if __name__ == '__main__': + + parser = argparse.ArgumentParser(description='Script to image replicas using WAL-E') + parser.add_argument('--scope', required=True) + parser.add_argument('--role', required=False) + parser.add_argument('--datadir', required=True) + parser.add_argument('--connstring', required=True) + parser.add_argument('--retries', type=int, default=1) + parser.add_argument('--envdir', required=True) + parser.add_argument('--threshold_megabytes', type=int, default=10240) + parser.add_argument('--threshold_backup_size_percentage', type=int, default=30) + parser.add_argument('--use_iam', type=int, default=0) + args = parser.parse_args() + + # retry cloning in a loop + for retry in range(0,args.retries + 1): + restore = WALERestore(scope=args.scope,datadir=args.datadir,connstring=args.constring, + env_dir=args.env_dir,threshold_mb=args.threshold_megabytes, + threshold_pct=args.threshold_backup_size_percentage) + restore.setup() + ret = restore.run() + if ret == 0: + break + + sys.exit(ret) diff --git a/postgres0.yml b/postgres0.yml index 41fca705..4d6f70fe 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -34,10 +34,6 @@ postgresql: data_dir: data/postgresql0 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True - pgpass: /tmp/pgpass0 - pg_rewind: - username: postgres - password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -46,16 +42,20 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: - username: postgres password: zalando admin: username: admin password: admin - wal_e: - env_dir: /home/postgres/etc/wal-e.d/env - threshold_megabytes: 10240 - threshold_backup_size_percentage: 30 create_replica_method: basebackup +# commented-out example for wal-e provisioning + #create_replica_method: wal_e, basebackup + #wal_e: + #command: /patroni/scripts/wale_restore.py + #env_dir: /home/postgres/etc/wal-e.d/env + #threshold_megabytes: 10240 + #threshold_backup_size_percentage: 30 + #retries: 2 + #use_iam: 1 #recovery_conf: #restore_command: cp ../wal_archive/%f %p parameters: @@ -67,4 +67,3 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" - wal_log_hints: "on" diff --git a/postgres1.yml b/postgres1.yml index 68597f2a..405fdc64 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -34,10 +34,6 @@ postgresql: data_dir: data/postgresql1 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True - pgpass: /tmp/pgpass1 - pg_rewind: - username: postgres - password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -46,18 +42,24 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: - user: postgres password: zalando admin: username: admin password: admin + create_replica_method: basebackup +# commented-out example for wal-e provisioning + #create_replica_method: wal_e, basebackup + #wal_e: + #command: /patroni/scripts/wale_restore.py + #env_dir: /home/postgres/etc/wal-e.d/env + #threshold_megabytes: 10240 + #threshold_backup_size_percentage: 30 + #retries: 2 + #use_iam: 1 + #recovery_conf: + #restore_command: cp ../wal_archive/%f %p #recovery_conf: #restore_command: cp ../wal_archive/%f %p - wal_e: - env_dir: /home/postgres/etc/wal-e.d/env - threshold_megabytes: 10240 - threshold_backup_size_percentage: 30 - create_replica_method: basebackup parameters: archive_mode: "on" wal_level: hot_standby @@ -67,4 +69,3 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" - wal_log_hints: "on" From e7a0ce57aa9084c510b471fb737a963d602a03e1 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Wed, 28 Oct 2015 14:55:42 -0700 Subject: [PATCH 08/25] Fixes, changes per discussion on pull request. Fixed logic path errors in postgresql.py. Cleaned up and shortened wale_restore.py. Reverted bad merge for YML files. --- patroni/postgresql.py | 9 ++-- patroni/scripts/wale_restore.py | 75 ++++++--------------------------- postgres0.yml | 11 +++-- postgres1.yml | 11 +++-- 4 files changed, 33 insertions(+), 73 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index d1083c26..5bbd95df 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -216,7 +216,7 @@ class Postgresql: ret = self.basebackup(leader, env) if ret == 0: # if basebackup succeeds, exit with success - return 0 + break else: # user-defined method; check for configuration # not required, actually @@ -247,17 +247,18 @@ class Postgresql: try: # call script with the full set of parameters - ret = subprocess.call(shlex.split(cmd) + shlex.split(method_config), env=env) + ret = subprocess.call(shlex.split(cmd) + params, env=env) # if we succeeded, stop if ret == 0: - return ret + break except Exception as e: logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, e.str)) ret = 1 # write the recovery.conf if ret == 0: - self.write_recovery_conf(leader) + ret = self.write_recovery_conf(leader) + return 0 # out of methods, return 1 return 1 diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index c645f75c..a448726c 100644 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -6,6 +6,8 @@ # note that pg_basebackup still expects to use restore from # WAL-E for transaction logs +# theoretically should work with SWIFT, but not tested on it + # arguments are: # - cluster scope # - cluster role @@ -18,9 +20,10 @@ # this script depends on an envdir defining the S3 bucket (or SWIFT dir),and login # credentials per WALE Documentation. -# DO NOT USE with additional restore_commands; this script writes the restore command +# currently also requires that you configure the restore_command to use wal_e, example: + #recovery_conf: + #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -# latest backup size import logging import os import psycopg2 @@ -35,7 +38,7 @@ if sys.hexversion >= 0x03000000: logger = logging.getLogger(__name__) -class Restore(object): +class WALERestore(object): def __init__(self, scope, role, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam): self.scope = scope @@ -49,59 +52,23 @@ class Restore(object): self.wal_e.iam_string = ' --aws-instance-profile ' else: self.wal_e.iam_string = '' - - def setup(self): - pass - - def replica_method(self): - return self.create_replica_with_pg_basebackup - - def replica_fallback_method(self): - return None - - def run(self): - """ creates a new replica using either pg_basebackup or WAL-E """ - method_fn = self.replica_method() - ret = method_fn() if method_fn else 1 - if ret != 0 and self.replica_fallback_method() is not None: - ret = (self.replica_fallback_method())() - return ret - - def create_replica_with_pg_basebackup(self): - try: - ret = subprocess.call(['pg_basebackup', '-R', '-D', '-x', - self.data_dir, '--host=' + self.master_connection['host'], - '--port=' + str(self.master_connection['port']), - '-U', self.master_connection['user']]) - except Exception as e: - logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e)) - return 1 - return ret - - -class WALERestore(Restore): - - def __init__(scope, role, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam): - super(WALERestore, self).__init__(scope, role, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam) - # check the environment variables - self.init_error = False - - def setup(self): - # check that we actually have an envdir if not os.path.exists(self.wal_e.dir): self.init_error = True else: + self.init_error = False self.wal_e.cmd = 'envdir {0} wal-e {1} '.\ format(self.wal_e.dir, self.wal_e.iam_string) + def run(self): + """ creates a new replica using WAL-E """ + ret = self.replica_method() + return ret + def replica_method(self): if self.should_use_s3_to_create_replica(): return self.create_replica_with_s3 return None - def replica_fallback_method(self): - return self.create_replica_with_pg_basebackup - def should_use_s3_to_create_replica(self): """ determine whether it makes sense to use S3 and not pg_basebackup """ if self.init_error: @@ -171,16 +138,6 @@ class WALERestore(Restore): # or exceeds the pre-determined size - pg_basebackup is chosen instead. return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\ (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100) - - def write_recovery_conf_wale(self): - restore_cmd = '{0} wal_fetch "%f" "%p"'.format(self.wal_e.cmd) - with open(os.path.join(self.data_dir, 'recovery.conf'), 'w') as f: - f.write("""standby_mode = 'on' -recovery_target_timeline = 'latest' -""") - f.write("""primary_conninfo = '{}'\n""".format(self.master_connection)) - f.write("""restore_command = '{}'\n""".format(restore_cmd)) - return 0 def create_replica_with_s3(self): if self.init_error: @@ -191,12 +148,7 @@ recovery_target_timeline = 'latest' except Exception as e: logger.error('Error when fetching backup with WAL-E: {0}'.format(e)) return 1 - - # if success, we need to write a recovery.conf for wal-e - # this doesn't work because we need data from main - #if ret == 0: - #ret = self.write_recovery_conf_wale() - + return ret @@ -219,7 +171,6 @@ if __name__ == '__main__': restore = WALERestore(scope=args.scope,datadir=args.datadir,connstring=args.constring, env_dir=args.env_dir,threshold_mb=args.threshold_megabytes, threshold_pct=args.threshold_backup_size_percentage) - restore.setup() ret = restore.run() if ret == 0: break diff --git a/postgres0.yml b/postgres0.yml index 4d6f70fe..1f9ace27 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -34,6 +34,10 @@ postgresql: data_dir: data/postgresql0 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True + pgpass: /tmp/pgpass0 + pg_rewind: + username: postgres + password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -42,22 +46,22 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: + username: postgres password: zalando admin: username: admin password: admin - create_replica_method: basebackup # commented-out example for wal-e provisioning #create_replica_method: wal_e, basebackup #wal_e: #command: /patroni/scripts/wale_restore.py - #env_dir: /home/postgres/etc/wal-e.d/env + #env_dir: /etc/wal-e.d/env #threshold_megabytes: 10240 #threshold_backup_size_percentage: 30 #retries: 2 #use_iam: 1 #recovery_conf: - #restore_command: cp ../wal_archive/%f %p + #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" parameters: archive_mode: "on" wal_level: hot_standby @@ -67,3 +71,4 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" + wal_log_hints: "on" diff --git a/postgres1.yml b/postgres1.yml index 405fdc64..fe1ec244 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -34,6 +34,10 @@ postgresql: data_dir: data/postgresql1 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True + pgpass: /tmp/pgpass1 + pg_rewind: + username: postgres + password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -42,11 +46,11 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: + user: postgres password: zalando admin: username: admin password: admin - create_replica_method: basebackup # commented-out example for wal-e provisioning #create_replica_method: wal_e, basebackup #wal_e: @@ -57,9 +61,7 @@ postgresql: #retries: 2 #use_iam: 1 #recovery_conf: - #restore_command: cp ../wal_archive/%f %p - #recovery_conf: - #restore_command: cp ../wal_archive/%f %p + #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" parameters: archive_mode: "on" wal_level: hot_standby @@ -69,3 +71,4 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" + wal_log_hints: "on" From 30aa83c5b27afc26a882f80a11dafba46e4b9c5f Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Mon, 2 Nov 2015 17:51:01 -0800 Subject: [PATCH 09/25] Fixed failing tests, pep8 issues. --- patroni/postgresql.py | 6 +++--- patroni/scripts/wale_restore.py | 20 ++++++++++---------- postgres0.yml | 1 + tests/test_postgresql.py | 4 ++-- 4 files changed, 16 insertions(+), 15 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 5bbd95df..728c555d 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -241,9 +241,9 @@ class Postgresql: else: cmd = replica_method method_config = {"scope": self.scope, - "role": "replica", - "datadir": self.data_dir, - "connstring": connstring} + "role": "replica", + "datadir": self.data_dir, + "connstring": connstring} try: # call script with the full set of parameters diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index a448726c..fca1f7d8 100644 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -3,7 +3,7 @@ # sample script to clone new replicas using WAL-E restore # falls back to pg_basebackup if WAL-E restore fails, or if # WAL-E backup is too far behind -# note that pg_basebackup still expects to use restore from +# note that pg_basebackup still expects to use restore from # WAL-E for transaction logs # theoretically should work with SWIFT, but not tested on it @@ -21,8 +21,8 @@ # credentials per WALE Documentation. # currently also requires that you configure the restore_command to use wal_e, example: - #recovery_conf: - #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" +# recovery_conf: +# restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" import logging import os @@ -137,8 +137,8 @@ class WALERestore(object): # if the size of the accumulated WAL segments is more than a certan percentage of the backup size # or exceeds the pre-determined size - pg_basebackup is chosen instead. return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\ - (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100) - + (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100) + def create_replica_with_s3(self): if self.init_error: return 1 @@ -148,7 +148,7 @@ class WALERestore(object): except Exception as e: logger.error('Error when fetching backup with WAL-E: {0}'.format(e)) return 1 - + return ret @@ -167,12 +167,12 @@ if __name__ == '__main__': args = parser.parse_args() # retry cloning in a loop - for retry in range(0,args.retries + 1): - restore = WALERestore(scope=args.scope,datadir=args.datadir,connstring=args.constring, - env_dir=args.env_dir,threshold_mb=args.threshold_megabytes, + for retry in range(0, args.retries + 1): + restore = WALERestore(scope=args.scope, datadir=args.datadir, connstring=args.constring, + env_dir=args.env_dir, threshold_mb=args.threshold_megabytes, threshold_pct=args.threshold_backup_size_percentage) ret = restore.run() if ret == 0: break - + sys.exit(ret) diff --git a/postgres0.yml b/postgres0.yml index 1f9ace27..40d246f0 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -51,6 +51,7 @@ postgresql: admin: username: admin password: admin + create_replica_method: basebackup # commented-out example for wal-e provisioning #create_replica_method: wal_e, basebackup #wal_e: diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 92f32a02..f198a28b 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -267,9 +267,9 @@ class TestPostgresql(unittest.TestCase): def test_create_replica(self): self.p.delete_trigger_file = Mock(side_effect=OSError()) with patch('subprocess.call', Mock(side_effect=[1, 0])): - self.assertEquals(self.p.create_replica({'host': '', 'port': '', 'user': ''}, ''), 0) + self.assertEquals(self.p.create_replica(self.leader, ''), 0) with patch('subprocess.call', Mock(side_effect=[Exception(), 0])): - self.assertEquals(self.p.create_replica({'host': '', 'port': '', 'user': ''}, ''), 0) + self.assertEquals(self.p.create_replica(self.leader, ''), 0) def test_create_connection_users(self): cfg = self.p.config From 7bc5ed7e4d470f37c5c7d5259f996489aa46746b Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Wed, 18 Nov 2015 18:03:38 +0100 Subject: [PATCH 10/25] Small fixes and an executable bit for wale script. --- patroni/scripts/wale_restore.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) mode change 100644 => 100755 patroni/scripts/wale_restore.py diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py old mode 100644 new mode 100755 index fca1f7d8..84ca3e83 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -24,6 +24,7 @@ # recovery_conf: # restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" +from collections import namedtuple import logging import os import psycopg2 @@ -40,11 +41,11 @@ logger = logging.getLogger(__name__) class WALERestore(object): - def __init__(self, scope, role, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam): + def __init__(self, scope, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam): self.scope = scope - self.role = role self.master_connection = connstring self.data_dir = datadir + self.wal_e = namedtuple('wale', 'dir,threshold_mb,threshold_pct,iam_string,cmd') self.wal_e.dir = env_dir self.wal_e.threshold_mb = threshold_mb self.wal_e.threshold_pct = threshold_pct @@ -168,9 +169,9 @@ if __name__ == '__main__': # retry cloning in a loop for retry in range(0, args.retries + 1): - restore = WALERestore(scope=args.scope, datadir=args.datadir, connstring=args.constring, - env_dir=args.env_dir, threshold_mb=args.threshold_megabytes, - threshold_pct=args.threshold_backup_size_percentage) + restore = WALERestore(scope=args.scope, datadir=args.datadir, connstring=args.connstring, + env_dir=args.envdir, threshold_mb=args.threshold_megabytes, + threshold_pct=args.threshold_backup_size_percentage, use_iam=args.use_iam) ret = restore.run() if ret == 0: break From d59ccd1d8e3e87ade5af49ecd757b927a7d6c5ed Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 20 Nov 2015 14:22:25 +0100 Subject: [PATCH 11/25] Fix a couple of logical issues. - command is deleted from method_config without checking whether it was there in the first place. - write_recovery_conf is called before the recovery file is restored from the backup location. --- patroni/postgresql.py | 26 ++++++++------------------ 1 file changed, 8 insertions(+), 18 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index c4c710f4..4ba9aab3 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -244,6 +244,7 @@ class Postgresql: replica_list = self.config.get('create_replica_method', 'basebackup') replica_methods = [rm.strip() for rm in replica_list.split(',')] # go through them in priority order + ret = 1 for replica_method in replica_methods: # if the method is basebackup, then use the built-in if replica_method == "basebackup": @@ -252,20 +253,17 @@ class Postgresql: # if basebackup succeeds, exit with success break else: + cmd = replica_method # user-defined method; check for configuration # not required, actually if replica_method in self.config: + method_config = self.config[replica_method].copy() # look to see if the user has supplied a full command path # if not, use the method name as the command - if "command" in self.config[replica_method]: - cmd = self.config[replica_method]["command"] - else: - cmd = replica_method - - # get the rest of the replica config - method_config = self.config[replica_method].copy() - # remove the command and turn it into a shlex set - del method_config["command"] + if "command" in method_config: + cmd = method_config["command"] + # remove the command and turn it into a shlex set + del method_config["command"] # add the default parameters method_config.update({"scope": self.scope, "role": "replica", @@ -273,12 +271,10 @@ class Postgresql: "connstring": connstring}) params = ["--{0}={1}".format(arg, val) for arg, val in method_config.items()] else: - cmd = replica_method method_config = {"scope": self.scope, "role": "replica", "datadir": self.data_dir, "connstring": connstring} - try: # call script with the full set of parameters ret = subprocess.call(shlex.split(cmd) + params, env=env) @@ -289,13 +285,7 @@ class Postgresql: logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, e.str)) ret = 1 - # write the recovery.conf - if ret == 0: - ret = self.write_recovery_conf(leader) - return 0 - - # out of methods, return 1 - return 1 + return ret def is_leader(self): return not self.query('SELECT pg_is_in_recovery()').fetchone()[0] From 4814e82055ed6d325fdf1baf66e87cd0311e6c6c Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 20 Nov 2015 17:55:51 +0100 Subject: [PATCH 12/25] Fix a typo and an error in calling subprocess.call. --- patroni/scripts/wale_restore.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index 84ca3e83..b1f7f10f 100755 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -36,6 +36,7 @@ import argparse if sys.hexversion >= 0x03000000: long = int +logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) logger = logging.getLogger(__name__) @@ -67,7 +68,7 @@ class WALERestore(object): def replica_method(self): if self.should_use_s3_to_create_replica(): - return self.create_replica_with_s3 + return self.create_replica_with_s3() return None def should_use_s3_to_create_replica(self): @@ -145,7 +146,7 @@ class WALERestore(object): return 1 # if we're set up, restore the replica using fetch latest try: - ret = subprocess.call(self.wal_e.cmd + ' backup-fetch {} LATEST'.format(self.data_dir)) + ret = subprocess.call(self.wal_e.cmd.split() + ['backup-fetch', '{}'.format(self.data_dir), 'LATEST']) except Exception as e: logger.error('Error when fetching backup with WAL-E: {0}'.format(e)) return 1 From 5370b46c6596565aecfd51fb9e7628bc619f5dfc Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 23 Nov 2015 15:27:15 +0100 Subject: [PATCH 13/25] Make sure WAL-E restore script returns 1 if criterias to use WAL-E were not met. --- patroni/scripts/wale_restore.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index b1f7f10f..03ce000e 100755 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -63,13 +63,9 @@ class WALERestore(object): def run(self): """ creates a new replica using WAL-E """ - ret = self.replica_method() - return ret - - def replica_method(self): if self.should_use_s3_to_create_replica(): return self.create_replica_with_s3() - return None + return 2 def should_use_s3_to_create_replica(self): """ determine whether it makes sense to use S3 and not pg_basebackup """ From 6c769554a3e7d3aeb20fa6a7a1c19a5c4dd116d7 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 23 Nov 2015 17:42:08 +0100 Subject: [PATCH 14/25] Add -p 1 to an example restore_command in order to limit WAL-E to only a single thread With out tests, the multi-threaded WAL fetch didn't work reliably with S3, resulting in stuck WAL-E processes. --- patroni/scripts/wale_restore.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index 03ce000e..4174a1b8 100755 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -22,7 +22,7 @@ # currently also requires that you configure the restore_command to use wal_e, example: # recovery_conf: -# restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" +# restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1 from collections import namedtuple import logging From 35efd36c5cae00cc5ff399b6c0779fac94953dd8 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 15:20:29 +0100 Subject: [PATCH 15/25] Improve unittests and make minor bugfixes. In particular, remove restore.py in favor of wale_restore.py, fix minor bugs in the latter and add unit tests. --- patroni/postgresql.py | 8 +- patroni/scripts/restore.py | 216 -------------------------------- patroni/scripts/wale_restore.py | 20 +-- tests/test_postgresql.py | 10 ++ tests/test_restore.py | 111 ---------------- tests/test_wale_restore.py | 127 +++++++++++++++++++ 6 files changed, 147 insertions(+), 345 deletions(-) delete mode 100755 patroni/scripts/restore.py delete mode 100644 tests/test_restore.py create mode 100644 tests/test_wale_restore.py diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 4ba9aab3..8a081b7d 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -229,8 +229,12 @@ class Postgresql: @staticmethod def build_connstring(conn): + """ + >>> Postgresql.build_connstring({'host': '127.0.0.1', 'port': '5432'}) == 'host=127.0.0.1 port=5432 ' + True + """ mconn = "" - for param, val in conn.items(): + for param, val in sorted(conn.items()): mconn = mconn + "{0}={1} ".format(param, val) return mconn @@ -282,7 +286,7 @@ class Postgresql: if ret == 0: break except Exception as e: - logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, e.str)) + logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, str(e))) ret = 1 return ret diff --git a/patroni/scripts/restore.py b/patroni/scripts/restore.py deleted file mode 100755 index 6b20e3e8..00000000 --- a/patroni/scripts/restore.py +++ /dev/null @@ -1,216 +0,0 @@ -#!/usr/bin/env python -# arguments are: -# - cluster scope -# - cluster role -# - master connection string - -# for the AWS, the folliowing environment variables should be defined: -# - WALE_ENV_DIR: directory where WAL-E environment is kept -# - WAL_S3_BUCKET: a name of the S3 bucket for WAL-E -# - WALE_BACKUP_THRESHOLD_MEGABYTES if WAL amount is above that - use pg_basebackup -# - WALE_BACKUP_THRESHOLD_PERCENTAGE if WAL size exceeds a certain percentage of the -# latest backup size -from collections import namedtuple -import logging -import os -import psycopg2 -import subprocess -import sys - - -if sys.hexversion >= 0x03000000: - long = int - -logger = logging.getLogger(__name__) - - -class Restore(object): - - def __init__(self, scope, role, datadir, connstring, env=None): - self.scope = scope - self.role = role - self.master_connection = Restore.parse_connstring(connstring) - self.data_dir = datadir - self.env = os.environ.copy() if not env else env - - @staticmethod - def parse_connstring(connstring): - # the connection string is in the form host= port= user= - # return the dictionary with all components as separare keys - result = {} - if connstring: - for x in connstring.split(): - if x and '=' in x: - key, val = x.split('=') - result[key.strip()] = val.strip() - return result - - def setup(self): - pass - - def replica_method(self): - return self.create_replica_with_pg_basebackup - - def replica_fallback_method(self): - return None - - def run(self): - """ creates a new replica using either pg_basebackup or WAL-E """ - method_fn = self.replica_method() - ret = method_fn() if method_fn else 1 - if ret != 0 and self.replica_fallback_method() is not None: - ret = (self.replica_fallback_method())() - return ret - - def create_replica_with_pg_basebackup(self): - try: - ret = subprocess.call(['pg_basebackup', '-R', '-D', - self.data_dir, '--host=' + self.master_connection['host'], - '--port=' + str(self.master_connection['port']), - '-U', self.master_connection['user']], - env=self.env) - except Exception as e: - logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e)) - return 1 - return ret - - -class WALERestore(Restore): - - def __init__(self, scope, role, datadir, connstring, env=None): - super(WALERestore, self).__init__(scope, role, datadir, connstring, env) - # check the environment variables - self.init_error = False - - def setup(self): - if (self.env.get('WAL_S3_BUCKET') and - self.env.get('WALE_BACKUP_THRESHOLD_PERCENTAGE') and - self.env.get('WALE_BACKUP_THRESHOLD_MEGABYTES')) is None: - self.init_error = True - else: - self.wal_e = namedtuple('WALE', - 'threshold_megabytes threshold_backup_size_percentage s3_bucket cmd dir env_file') - - self.wal_e.dir = self.env.get('WALE_ENV_DIR', '/home/postgres/etc/wal-e.d/env') - self.wal_e.env_file = os.path.join(self.wal_e.dir, 'WALE_S3_PREFIX') - - self.wal_e.cmd = 'envdir {} wal-e --aws-instance-profile '.\ - format(self.wal_e.dir) - self.wal_e.s3_bucket = self.env['WAL_S3_BUCKET'] - self.wal_e.threshold_megabytes = self.env['WALE_BACKUP_THRESHOLD_MEGABYTES'] - self.wal_e.threshold_backup_size_percentage = self.env['WALE_BACKUP_THRESHOLD_PERCENTAGE'] - - # check that the env file exists, create it otherwise - try: - if not os.path.exists(self.wal_e.dir): - os.makedirs(self.wal_e.dir) - # if this is a directory - make sure we have full access there - elif not (os.path.isdir(self.wal_e.dir) and os.access(self.wal_e.dir, os.R_OK | os.W_OK | os.X_OK)): - logger.error("Unable to access {} or not a directory".format(self.wal_e.dir)) - self.init_error = True - # if WAL_S3_PREFIX is not there - create it and write the full path to bucket - if not self.init_error and not os.path.exists(self.wal_e.env_file): - with open(self.wal_e.env_file, 'w') as f: - f.write("s3://{0}/spilo/{1}/wal/\n".format(self.wal_e.s3_bucket, self.scope)) - - except (os.error, IOError) as e: - logger.error("{0}: WAL-e archiving is disabled".format(e)) - self.init_error = True - - def replica_method(self): - if self.should_use_s3_to_create_replica(): - return self.create_replica_with_s3 - return None - - def replica_fallback_method(self): - return self.create_replica_with_pg_basebackup - - def should_use_s3_to_create_replica(self): - """ determine whether it makes sense to use S3 and not pg_basebackup """ - if self.init_error: - return False - - threshold_megabytes = self.wal_e.threshold_megabytes - threshold_backup_size_percentage = self.wal_e.threshold_backup_size_percentage - - try: - latest_backup = subprocess.check_output(self.wal_e.cmd.split() + ['backup-list', '--detail', 'LATEST'], - env=self.env) - # name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start - # wal_segment_backup_stop wal_segment_offset_backup_stop - # base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z - # 20310671 00000001000000000000007F 00000040 - # 00000001000000000000007F 00000240 - backup_strings = latest_backup.splitlines() if latest_backup else () - if len(backup_strings) != 2: - return False - - names = backup_strings[0].split() - vals = backup_strings[1].split() - if (len(names) != len(vals)) or (len(names) != 7): - return False - - backup_info = dict(zip(names, vals)) - except subprocess.CalledProcessError as e: - logger.error("could not query wal-e latest backup: {}".format(e)) - return False - - try: - backup_size = backup_info['expanded_size_bytes'] - backup_start_segment = backup_info['wal_segment_backup_start'] - backup_start_offset = backup_info['wal_segment_offset_backup_start'] - except Exception as e: - logger.error("unable to get some of S3 backup parameters: {}".format(e)) - return False - - # WAL filename is XXXXXXXXYYYYYYYY000000ZZ, where X - timeline, Y - LSN logical log file, - # ZZ - 2 high digits of LSN offset. The rest of the offset is the provided decimal offset, - # that we have to convert to hex and 'prepend' to the high offset digits. - - lsn_segment = backup_start_segment[8:16] - # first 2 characters of the result are 0x and the last one is L - lsn_offset = hex((long(backup_start_segment[16:32], 16) << 24) + long(backup_start_offset))[2:-1] - - # construct the LSN from the segment and offset - backup_start_lsn = '{}/{}'.format(lsn_segment, lsn_offset) - - conn = None - cursor = None - diff_in_bytes = long(backup_size) - try: - # get the difference in bytes between the current WAL location and the backup start offset - conn = psycopg2.connect(**(self.master_connection)) - conn.autocommit = True - cursor = conn.cursor() - cursor.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,)) - diff_in_bytes = long(cursor.fetchone()[0]) - except psycopg2.Error as e: - logger.error('could not determine difference with the master location: {}'.format(e)) - return False - finally: - cursor and cursor.close() - conn and conn.close() - - # if the size of the accumulated WAL segments is more than a certan percentage of the backup size - # or exceeds the pre-determined size - pg_basebackup is chosen instead. - return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\ - (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100) - - def create_replica_with_s3(self): - if self.init_error: - return 1 - try: - ret = subprocess.call(self.wal_e.cmd + ' backup-fetch {} LATEST'.format(self.data_dir), env=self.env) - except Exception as e: - logger.error('Error when fetching backup with WAL-E: {0}'.format(e)) - return 1 - return ret - - -if __name__ == '__main__': - if len(sys.argv) == 5: - # scope, role, datadir, connstring - restore = WALERestore(*(sys.argv[1:])) - restore.setup() - sys.exit(restore.run()) - sys.exit("Usage: {0} scope role datadir connstring".format(sys.argv[0])) diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index 03ce000e..ec9a023f 100755 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -50,27 +50,18 @@ class WALERestore(object): self.wal_e.dir = env_dir self.wal_e.threshold_mb = threshold_mb self.wal_e.threshold_pct = threshold_pct - if use_iam == 1: - self.wal_e.iam_string = ' --aws-instance-profile ' - else: - self.wal_e.iam_string = '' - if not os.path.exists(self.wal_e.dir): - self.init_error = True - else: - self.init_error = False - self.wal_e.cmd = 'envdir {0} wal-e {1} '.\ - format(self.wal_e.dir, self.wal_e.iam_string) + self.wal_e.iam_string = ' --aws-instance-profile ' if use_iam == 1 else '' + self.wal_e.cmd = 'envdir {0} wal-e {1} '.format(self.wal_e.dir, self.wal_e.iam_string) + self.init_error = (not os.path.exists(self.wal_e.dir)) def run(self): """ creates a new replica using WAL-E """ - if self.should_use_s3_to_create_replica(): + if not self.init_error and self.should_use_s3_to_create_replica(): return self.create_replica_with_s3() return 2 def should_use_s3_to_create_replica(self): """ determine whether it makes sense to use S3 and not pg_basebackup """ - if self.init_error: - return False threshold_megabytes = self.wal_e.threshold_mb threshold_backup_size_percentage = self.wal_e.threshold_pct @@ -138,8 +129,6 @@ class WALERestore(object): (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100) def create_replica_with_s3(self): - if self.init_error: - return 1 # if we're set up, restore the replica using fetch latest try: ret = subprocess.call(self.wal_e.cmd.split() + ['backup-fetch', '{}'.format(self.data_dir), 'LATEST']) @@ -151,7 +140,6 @@ class WALERestore(object): if __name__ == '__main__': - parser = argparse.ArgumentParser(description='Script to image replicas using WAL-E') parser.add_argument('--scope', required=True) parser.add_argument('--role', required=False) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index ba108c28..e13e65fa 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -284,6 +284,16 @@ class TestPostgresql(unittest.TestCase): with patch('subprocess.call', Mock(side_effect=[Exception(), 0])): self.assertEquals(self.p.create_replica(self.leader, ''), 0) + self.p.config['create_replica_method'] = 'wale, basebackup' + self.p.config['wale'] = {'command': 'foo'} + with patch('subprocess.call', Mock(return_value=0)): + self.assertEquals(self.p.create_replica(self.leader, ''), 0) + del self.p.config['wale'] + self.assertEquals(self.p.create_replica(self.leader, ''), 0) + + with patch('subprocess.call', Mock(side_effect=Exception("foo"))): + self.assertEquals(self.p.create_replica(self.leader, ''), 1) + def test_create_connection_users(self): cfg = self.p.config cfg['superuser']['username'] = 'test' diff --git a/tests/test_restore.py b/tests/test_restore.py deleted file mode 100644 index 2ffd8a58..00000000 --- a/tests/test_restore.py +++ /dev/null @@ -1,111 +0,0 @@ -import unittest -from mock import MagicMock, patch -import os -from patroni.scripts.restore import Restore, WALERestore - - -def fake_cursor_fetchone(*args, **kwargs): - return ('16777216',) - - -def fake_call_fail_for_wal_e(*args, **kwargs): - if len(args) > 0 and 'backup-fetch' in args[0]: - return 1 - return 0 - - -def fake_call_fail_for_base_backup(*args, **kwargs): - if len(args) > 0 and 'backup-fetch' in args[0]: - return 0 - return 1 - - -def fake_backup_data(self, *args, **kwargs): - """ return the fake result of WAL-E backup-list""" - return """name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop -base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 00000001000000000000007F 00000040 00000001000000000000007F 00000240 -""" - - -class TestRestore(unittest.TestCase): - - def setUp(self): - self.restore = Restore("batman", "master", "/data", "host=batman port=5432 user=batman") - pass - - def tearDown(self): - pass - - def test_parse_connstring(self): - self.assertDictEqual(self.restore.master_connection, {'host': 'batman', 'port': '5432', 'user': 'batman'}) - - @patch('subprocess.call', MagicMock(return_value=0)) - def test_run(self): - ret = self.restore.run() - self.assertEqual(ret, 0) - - @patch('subprocess.call', MagicMock(return_value=1)) - def test_run_fail(self): - ret = self.restore.run() - self.assertEqual(ret, 1) - - -@patch('os.access', MagicMock(return_value=True)) -@patch('os.makedirs', MagicMock(return_value=True)) -@patch('os.path.exists', MagicMock(return_value=True)) -@patch('os.path.isdir', MagicMock(return_value=True)) -@patch('psycopg2.extensions.cursor.fetchone', MagicMock(side_effect=fake_cursor_fetchone)) -@patch('psycopg2.extensions.cursor', MagicMock(autospec=True)) -@patch('psycopg2.extensions.connection', MagicMock(autospec=True)) -@patch('psycopg2.connect', MagicMock(autospec=True)) -@patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data)) -class TestWALERestore(unittest.TestCase): - - def setUp(self): - env = {} - env['WAL_S3_BUCKET'] = 'batman' - env['WALE_BACKUP_THRESHOLD_PERCENTAGE'] = 100 - env['WALE_BACKUP_THRESHOLD_MEGABYTES'] = 100 - self.wale_restore = WALERestore("batman", "master", "/data", "host=batman port=5432 user=batman", env=env) - - def tearDown(self): - pass - - def test_setup(self): - self.wale_restore.setup() - self.assertFalse(self.wale_restore.init_error) - - # have to redefine the class-level os.access mock inside the function - # since the class-level mock will be applied after the function level one. - @patch('os.access', return_value=False) - def test_setup_fail(self, mock_no_access): - os.access = mock_no_access - self.wale_restore.setup() - self.assertTrue(self.wale_restore.init_error) - - # The 3 tests above only differ with the mock function instead of a subprocess call - # in the first one, subprocess call should return success only for wal-e command, - # checking the primary use-case of restoring from WAL-E backup. - # In the second one, we test fallbacks by failing at WAL-E, but succeeding at - # pg_basebackup. - # Finally, the last use case is when all subprocess.call fails. resulting in a - # failure to restore from replica - @patch('subprocess.call', - MagicMock(side_effect=lambda *args, **kwargs: 0 if 'wal-e' in args[0] else 1)) - def test_run(self): - self.wale_restore.setup() - ret = self.wale_restore.run() - self.assertEqual(ret, 0) - - @patch('subprocess.call', - MagicMock(side_effect=lambda *args, **kwargs: 0 if 'pg_basebackup' in args[0] else 1)) - def test_run_fallback(self): - self.wale_restore.setup() - ret = self.wale_restore.run() - self.assertEqual(ret, 0) - - @patch('subprocess.call', MagicMock(return_value=1)) - def test_run_all_fail(self): - self.wale_restore.setup() - ret = self.wale_restore.run() - self.assertEqual(ret, 1) diff --git a/tests/test_wale_restore.py b/tests/test_wale_restore.py new file mode 100644 index 00000000..747fa5c2 --- /dev/null +++ b/tests/test_wale_restore.py @@ -0,0 +1,127 @@ +import unittest +from mock import MagicMock, patch, PropertyMock +import os +import psycopg2 +import subprocess +from patroni.scripts.wale_restore import WALERestore + + +def fake_cursor_fetchone(*args, **kwargs): + return ('16777216',) + + +def fake_call_fail_for_wal_e(*args, **kwargs): + if len(args) > 0 and 'backup-fetch' in args[0]: + return 1 + return 0 + + +def fake_call_fail_for_base_backup(*args, **kwargs): + if len(args) > 0 and 'backup-fetch' in args[0]: + return 0 + return 1 + + +def fake_backup_data(self, *args, **kwargs): + """ return the fake result of WAL-E backup-list""" + return """name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop +base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 00000001000000000000007F 00000040 00000001000000000000007F 00000240 +""" + +def fake_backup_data_2(self, *args, **kwargs): + """ return the fake result of WAL-E backup-list""" + return """name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop """ + +def fake_backup_data_3(self, *args, **kwargs): + """ return the fake result of WAL-E backup-list""" + return """name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop +base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 00000001000000000000007F 00000040 00000001000000000000007F 00000240 +""" + +def fake_backup_data_4(self, *args, **kwargs): + """ return the fake result of WAL-E backup-list""" + return """name last_modified expanded_size_foo wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop +base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 00000001000000000000007F 00000040 00000001000000000000007F 00000240 +""" + + +@patch('os.access', MagicMock(return_value=True)) +@patch('os.makedirs', MagicMock(return_value=True)) +@patch('os.path.exists', MagicMock(return_value=True)) +@patch('os.path.isdir', MagicMock(return_value=True)) +@patch('psycopg2.extensions.cursor.fetchone', MagicMock(side_effect=fake_cursor_fetchone)) +@patch('psycopg2.extensions.cursor', MagicMock(autospec=True)) +@patch('psycopg2.extensions.connection', MagicMock(autospec=True)) +@patch('psycopg2.connect', MagicMock(autospec=True)) +@patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data)) +class TestWALERestore(unittest.TestCase): + + def setUp(self): + self.wale_restore = WALERestore("batman", "/data", + "host=batman port=5432 user=batman", "/etc", 100, 100, 1) + + def tearDown(self): + pass + + def test_should_use_s3_to_create_replica(self): + with patch('psycopg2.connect', MagicMock(side_effect=psycopg2.Error("foo"))): + self.assertFalse(self.wale_restore.should_use_s3_to_create_replica()) + with patch('subprocess.check_output', MagicMock(side_effect=subprocess.CalledProcessError(1, "cmd", "foo"))): + self.assertFalse(self.wale_restore.should_use_s3_to_create_replica()) + with patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data_2)): + self.assertFalse(self.wale_restore.should_use_s3_to_create_replica()) + with patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data_3)): + self.assertFalse(self.wale_restore.should_use_s3_to_create_replica()) + with patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data_4)): + self.assertFalse(self.wale_restore.should_use_s3_to_create_replica()) + + self.wale_restore.should_use_s3_to_create_replica() + + def test_create_replica_with_s3(self): + with patch('subprocess.call', MagicMock(return_value=0)): + self.assertEqual(self.wale_restore.create_replica_with_s3(), 0) + with patch('subprocess.call', MagicMock(side_effect=Exception("foo"))): + self.assertEqual(self.wale_restore.create_replica_with_s3(), 1) + + def test_run(self): + with patch.object(self.wale_restore, 'init_error', PropertyMock(return_value=True)): + self.assertEqual(self.wale_restore.run(), 2) + with patch.object(self.wale_restore, 'should_use_s3_to_create_replica', MagicMock(return_value=True)): + with patch.object(self.wale_restore, 'create_replica_with_s3', MagicMock(return_value=0)): + self.assertEqual(self.wale_restore.run(), 0) + + # with patch.object(self.wale_restore, 'init_error', PropertyMock(return_value=True)): + # self.assertFalse(self.wale_restore.create_replica_with_s3()) + + # @patch('subprocess.call', MagicMock(return_value=0)) + # def test_run(self): + # ret = self.wale_restore.run() + # self.assertEqual(ret, 0) + + + # The 3 tests above only differ with the mock function instead of a subprocess call + # in the first one, subprocess call should return success only for wal-e command, + # checking the primary use-case of restoring from WAL-E backup. + # In the second one, we test fallbacks by failing at WAL-E, but succeeding at + # pg_basebackup. + # Finally, the last use case is when all subprocess.call fails. resulting in a + # failure to restore from replica + # @patch('subprocess.call', + # MagicMock(side_effect=lambda *args, **kwargs: 0 if 'wal-e' in args[0] else 1)) + # def test_run(self): + # self.wale_restore.setup() + # ret = self.wale_restore.run() + # self.assertEqual(ret, 0) + + # @patch('subprocess.call', + # MagicMock(side_effect=lambda *args, **kwargs: 0 if 'pg_basebackup' in args[0] else 1)) + # def test_run_fallback(self): + # self.wale_restore.setup() + # ret = self.wale_restore.run() + # self.assertEqual(ret, 0) + + # @patch('subprocess.call', MagicMock(return_value=1)) + # def test_run_all_fail(self): + # self.wale_restore.setup() + # ret = self.wale_restore.run() + # self.assertEqual(ret, 1) From f3d9edb57fff02dfca413bf49e348b3b475739d0 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 15:32:44 +0100 Subject: [PATCH 16/25] also add -p 1 to the restore commands provided with sample yaml files. --- postgres0.yml | 2 +- postgres1.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/postgres0.yml b/postgres0.yml index ae3646cf..66690367 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -79,7 +79,7 @@ postgresql: #retries: 2 #use_iam: 1 #recovery_conf: - #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" + #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1 parameters: archive_mode: "on" wal_level: hot_standby diff --git a/postgres1.yml b/postgres1.yml index d96df08c..56072a52 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -78,7 +78,7 @@ postgresql: #retries: 2 #use_iam: 1 #recovery_conf: - #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" + #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1 parameters: archive_mode: "on" wal_level: hot_standby From fcbb820949b7bd4b75b2f52affc122a79155fc4c Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 15:43:56 +0100 Subject: [PATCH 17/25] Remove some cruft from the tests. --- tests/test_wale_restore.py | 36 ------------------------------------ 1 file changed, 36 deletions(-) diff --git a/tests/test_wale_restore.py b/tests/test_wale_restore.py index 747fa5c2..05f34187 100644 --- a/tests/test_wale_restore.py +++ b/tests/test_wale_restore.py @@ -89,39 +89,3 @@ class TestWALERestore(unittest.TestCase): with patch.object(self.wale_restore, 'should_use_s3_to_create_replica', MagicMock(return_value=True)): with patch.object(self.wale_restore, 'create_replica_with_s3', MagicMock(return_value=0)): self.assertEqual(self.wale_restore.run(), 0) - - # with patch.object(self.wale_restore, 'init_error', PropertyMock(return_value=True)): - # self.assertFalse(self.wale_restore.create_replica_with_s3()) - - # @patch('subprocess.call', MagicMock(return_value=0)) - # def test_run(self): - # ret = self.wale_restore.run() - # self.assertEqual(ret, 0) - - - # The 3 tests above only differ with the mock function instead of a subprocess call - # in the first one, subprocess call should return success only for wal-e command, - # checking the primary use-case of restoring from WAL-E backup. - # In the second one, we test fallbacks by failing at WAL-E, but succeeding at - # pg_basebackup. - # Finally, the last use case is when all subprocess.call fails. resulting in a - # failure to restore from replica - # @patch('subprocess.call', - # MagicMock(side_effect=lambda *args, **kwargs: 0 if 'wal-e' in args[0] else 1)) - # def test_run(self): - # self.wale_restore.setup() - # ret = self.wale_restore.run() - # self.assertEqual(ret, 0) - - # @patch('subprocess.call', - # MagicMock(side_effect=lambda *args, **kwargs: 0 if 'pg_basebackup' in args[0] else 1)) - # def test_run_fallback(self): - # self.wale_restore.setup() - # ret = self.wale_restore.run() - # self.assertEqual(ret, 0) - - # @patch('subprocess.call', MagicMock(return_value=1)) - # def test_run_all_fail(self): - # self.wale_restore.setup() - # ret = self.wale_restore.run() - # self.assertEqual(ret, 1) From 875c82e833fa50ab11bd7179e9293572a682b558 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 24 Nov 2015 16:02:46 +0100 Subject: [PATCH 18/25] Documentation bugfix --- README.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.rst b/README.rst index f752c2ca..422e0849 100644 --- a/README.rst +++ b/README.rst @@ -113,11 +113,11 @@ For an example file, see ``postgres0.yml``. Regarding settings: - *recovery\_conf*: additional configuration settings written to recovery.conf when configuring follower. - *parameters*: list of configuration settings for Postgres. Many of these are required for replication to work. - - *create_replica_methods*: an ordered list of the create methods for turning a patroni node into a new replica. + - *create\_replica\_methods*: an ordered list of the create methods for turning a patroni node into a new replica. "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured as its own config item. - - *{replica_method}* for each create_replica_method other than basebackup, you would add a configuration section + - *replica\_method* for each create_replica_method other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed. Other configuration parameters will be passed along to the script in the form "parameter=value". From aee7d32af677882a71ae9ca49a4360855f1b4868 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 16:14:39 +0100 Subject: [PATCH 19/25] Small code improvement, per comment by Alex. --- patroni/postgresql.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 8a081b7d..b41d0fff 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -264,10 +264,7 @@ class Postgresql: method_config = self.config[replica_method].copy() # look to see if the user has supplied a full command path # if not, use the method name as the command - if "command" in method_config: - cmd = method_config["command"] - # remove the command and turn it into a shlex set - del method_config["command"] + cmd = method_config.pop('command', cmd) # add the default parameters method_config.update({"scope": self.scope, "role": "replica", From 6d296b1b347415d81ad8cd143ab074663ba50f89 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 16:26:08 +0100 Subject: [PATCH 20/25] Make sure params passed to the replica creation method are always defined. Per code review by Alex. --- patroni/postgresql.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index b41d0fff..932634ed 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -258,6 +258,7 @@ class Postgresql: break else: cmd = replica_method + method_config = {} # user-defined method; check for configuration # not required, actually if replica_method in self.config: @@ -266,17 +267,12 @@ class Postgresql: # if not, use the method name as the command cmd = method_config.pop('command', cmd) # add the default parameters + try: method_config.update({"scope": self.scope, "role": "replica", "datadir": self.data_dir, "connstring": connstring}) params = ["--{0}={1}".format(arg, val) for arg, val in method_config.items()] - else: - method_config = {"scope": self.scope, - "role": "replica", - "datadir": self.data_dir, - "connstring": connstring} - try: # call script with the full set of parameters ret = subprocess.call(shlex.split(cmd) + params, env=env) # if we succeeded, stop From 4b1ff5a4bb6319ebd3ad579b28dbc9adb078e083 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 16:37:40 +0100 Subject: [PATCH 21/25] Code refactoring, per code review by Alex. --- patroni/postgresql.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 932634ed..47fd53d3 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -714,7 +714,7 @@ $$""".format(name, options), name, password, password) bbfailures = 0 maxfailures = 2 ret = 1 - while bbfailures < maxfailures: + for bbfailures in range(0, maxfailures): try: ret = subprocess.call(['pg_basebackup', '--pgdata=' + self.data_dir, '--xlog-method=stream', "--dbname=" + master_connection], env=env) @@ -724,9 +724,8 @@ $$""".format(name, options), name, password, password) except Exception as e: logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e)) - bbfailures += 1 - if bbfailures < maxfailures: + if bbfailures < maxfailures - 1: logger.error('Trying again in 5 seconds') - time.sleep(5) + time.sleep(5) return ret From d8a8fe9a804d3214a6f88c2ae2a9dc92e2bf6c25 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 16:40:18 +0100 Subject: [PATCH 22/25] Convert build_connstring into a one-liner, per code review by Alex. --- patroni/postgresql.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 47fd53d3..a760e879 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -230,14 +230,10 @@ class Postgresql: @staticmethod def build_connstring(conn): """ - >>> Postgresql.build_connstring({'host': '127.0.0.1', 'port': '5432'}) == 'host=127.0.0.1 port=5432 ' + >>> Postgresql.build_connstring({'host': '127.0.0.1', 'port': '5432'}) == 'host=127.0.0.1 port=5432' True """ - mconn = "" - for param, val in sorted(conn.items()): - mconn = mconn + "{0}={1} ".format(param, val) - - return mconn + return ' '.join('{}={}'.format(param, val) for param, val in sorted(conn.items())) def create_replica(self, leader, env): # create the replica according to the replica_method From be9e525739a8f53fe0e18d8e215c9b7166592248 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 24 Nov 2015 17:23:38 +0100 Subject: [PATCH 23/25] Remove an unused line. --- patroni/postgresql.py | 1 - 1 file changed, 1 deletion(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index a760e879..37c20764 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -707,7 +707,6 @@ $$""".format(name, options), name, password, password) # tries twice, then returns failure (as 1) # uses "stream" as the xlog-method to avoid sync issues master_connection = leader.conn_url - bbfailures = 0 maxfailures = 2 ret = 1 for bbfailures in range(0, maxfailures): From 14b8dfa3e8b5f43e3640b2872c79c24a35902661 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Wed, 25 Nov 2015 10:29:17 +0100 Subject: [PATCH 24/25] Make create_replica_method a YAML array. Make sure the absense of this key or empty value in it is handled correctly. Update tests and sample configuration files. --- patroni/postgresql.py | 6 +++--- postgres0.yml | 4 +++- postgres1.yml | 5 ++++- tests/test_postgresql.py | 2 +- 4 files changed, 11 insertions(+), 6 deletions(-) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 37c20764..033de46a 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -240,9 +240,9 @@ class Postgresql: # defined by the user. this is a list, so we need to # loop through all methods the user supplies connstring = leader.conn_url - # get list of replica methods from config - replica_list = self.config.get('create_replica_method', 'basebackup') - replica_methods = [rm.strip() for rm in replica_list.split(',')] + # get list of replica methods from config. + # If there is no configuration key, or no value is specified, use basebackup + replica_methods = self.config.get('create_replica_method') or ['basebackup'] # go through them in priority order ret = 1 for replica_method in replica_methods: diff --git a/postgres0.yml b/postgres0.yml index 66690367..84c74acd 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -68,7 +68,9 @@ postgresql: admin: username: admin password: admin - create_replica_method: basebackup + create_replica_method: + - basebackup +# - wal_e # commented-out example for wal-e provisioning #create_replica_method: wal_e, basebackup #wal_e: diff --git a/postgres1.yml b/postgres1.yml index 56072a52..32919843 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -69,7 +69,10 @@ postgresql: username: admin password: admin # commented-out example for wal-e provisioning - #create_replica_method: wal_e, basebackup + create_replica_method: + - basebackup +# - wal_e +# commented-out example for wal-e provisioning #wal_e: #command: /patroni/scripts/wale_restore.py #env_dir: /home/postgres/etc/wal-e.d/env diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index e13e65fa..2097bcb7 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -284,7 +284,7 @@ class TestPostgresql(unittest.TestCase): with patch('subprocess.call', Mock(side_effect=[Exception(), 0])): self.assertEquals(self.p.create_replica(self.leader, ''), 0) - self.p.config['create_replica_method'] = 'wale, basebackup' + self.p.config['create_replica_method'] = ['wale', 'basebackup'] self.p.config['wale'] = {'command': 'foo'} with patch('subprocess.call', Mock(return_value=0)): self.assertEquals(self.p.create_replica(self.leader, ''), 0) From d4ab4d1aef9bbee13c3efe498e36cba5b52a11b1 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Wed, 25 Nov 2015 14:48:59 +0100 Subject: [PATCH 25/25] Output the method used to initialize the replica. --- patroni/postgresql.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 033de46a..fba9a22e 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -250,6 +250,7 @@ class Postgresql: if replica_method == "basebackup": ret = self.basebackup(leader, env) if ret == 0: + logger.info("replica has been created using basebackup") # if basebackup succeeds, exit with success break else: @@ -273,6 +274,7 @@ class Postgresql: ret = subprocess.call(shlex.split(cmd) + params, env=env) # if we succeeded, stop if ret == 0: + logger.info("replica has been created using {0}".format(replica_method)) break except Exception as e: logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, str(e)))