From fa7d36da9b1984f0cbe94299c13da58ba2911cba Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 22 Oct 2015 17:21:39 -0700 Subject: [PATCH 1/9] 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 2/9] 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 3/9] 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 4/9] 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 5/9] 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 6/9] 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 7/9] 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 8/9] 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 9/9] 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