From 7f2dcc0a955bc0e6ae460da5a49c3917650aa79a Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 3 Aug 2015 11:44:11 +0200 Subject: [PATCH 01/13] Move replica creation code into a separte file. So far, only the class that makes replicas with pg_basebackup is called. --- helpers/postgresql.py | 97 ++----------------------- postgres0.yml | 1 + scripts/restore.py | 153 +++++++++++++++++++++++++++++++++++++++ tests/test_postgresql.py | 3 +- 4 files changed, 163 insertions(+), 91 deletions(-) create mode 100644 scripts/restore.py diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 84a94346..a7810573 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -145,99 +145,16 @@ class Postgresql: env['PGPASSFILE'] = pgpass return self.create_replica(r, env) == 0 + @staticmethod + def build_connstring(self, conn): + return "host={host} port={port} user={user}".format(**conn) + def create_replica(self, master_connection, env): - """ creates a new replica using either pg_basebackup or WAL-E """ - if self.should_use_s3_to_create_replica(master_connection): - result = self.create_replica_with_s3() - # if restore from the backup on S3 failed - try with the pg_basebackup - if result == 0: - return result - return self.create_replica_with_pg_basebackup(master_connection, env) - - def create_replica_with_pg_basebackup(self, master_connection, env): - ret = subprocess.call(['pg_basebackup', '-R', '-D', self.data_dir, '--host=' + master_connection['host'], - '--port=' + str(master_connection['port']), '-U', master_connection['user']], env=env) - self.delete_trigger_file() + connstring = self.build_connstring(master_connection, master_connection) + cmd = self.config['restore'] + ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", self.data_dir, connstring], env=env) return ret - def create_replica_with_s3(self): - if not self.wal_e or not self.wal_e_path: - return 1 - - ret = subprocess.call(self.wal_e_path + ' backup-fetch {} LATEST'.format(self.data_dir), shell=True) - self.restore_configuration_files() - return ret - - def should_use_s3_to_create_replica(self, master_connection): - """ determine whether it makes sense to use S3 and not pg_basebackup """ - if not self.wal_e or not self.wal_e_path: - return False - - threshold_megabytes = self.wal_e.get('threshold_megabytes', 10240) - threshold_backup_size_percentage = self.wal_e.get('threshold_backup_size_percentage', 30) - - try: - latest_backup = subprocess.check_output(self.wal_e_path.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 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(**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 is_leader(self, check_only=False): ret = not self.query('SELECT pg_is_in_recovery()').fetchone()[0] if ret and self.is_promoted and not check_only: diff --git a/postgres0.yml b/postgres0.yml index 293ebbc5..200792d5 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -46,6 +46,7 @@ postgresql: env_dir: /home/postgres/etc/wal-e.d/env threshold_megabytes: 10240 threshold_backup_size_percentage: 30 + restore: /usr/bin/true #recovery_conf: #restore_command: cp ../wal_archive/%f %p parameters: diff --git a/scripts/restore.py b/scripts/restore.py new file mode 100644 index 00000000..f035c794 --- /dev/null +++ b/scripts/restore.py @@ -0,0 +1,153 @@ +# arguments are: +# - cluster scope +# - cluster role +# - master connection string +import logging +import os +import psycopg2 +import subprocess +import sys + +logger = logging.getLogger(__name__) + + +class Restore: + + def __init__(self, scope, role, datadir, connstring): + self.scope = scope + self.role = role + self.connstring = connstring + self.datadir = datadir + self.env = os.environ.copy() + + def parse_connstring(self): + # the connection string is in the form host= port= user= + # return the dictionary with all components as separare keys + result = {} + if self.connstring: + for x in self.connstring.split(): + if x and '=' in x: + key, val = x.split('=') + result[key.strip()] = val.strip() + return result + + 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 ret != 0 and self.replica_fallback_method() is not None: + ret = (self.replica_fallback_method())() + return ret + + def create_replica_with_pg_basebackup(self): + master_connection = self.parse_connstring() + ret = subprocess.call(['pg_basebackup', '-R', '-D', self.data_dir, '--host=' + master_connection['host'], + '--port=' + str(master_connection['port']), '-U', master_connection['user']], env=self.env) + self.delete_trigger_file() + return ret + + +class WALERestore(Restore): + + def __init__(self, scope, role, datadir, connstring): + super(WALERestore, self).__init__(scope, role, datadir, connstring) + + def replica_method(self): + if self.should_use_s3_to_create_replica(self): + return self.create_replica_with_s3 + return self.create_replica_with_pg_basebackup + + def replica_fallback_method(self): + return self.create_replica_with_pg_basebackup + + def should_use_s3_to_create_replica(self, master_connection): + """ determine whether it makes sense to use S3 and not pg_basebackup """ + if not self.wal_e or not self.wal_e_paselfth: + return False + + threshold_megabytes = self.wal_e.get('threshold_megabytes', 10240) + threshold_backup_size_percentage = self.wal_e.get('threshold_backup_size_percentage', 30) + + try: + latest_backup = subprocess.check_output(self.wal_e_path.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 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(**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 not self.wal_e or not self.wal_e_path: + return 1 + + ret = subprocess.call(self.wal_e_path + ' backup-fetch {} LATEST'.format(self.data_dir), shell=True) + self.restore_configuration_files() + return ret + + +if __name__ == '__main__': + if len(sys.argv) == 5: + # scope, role, datadir, connstring + restore = Restore(*(sys.argv[1:])) + sys.exit(restore.run()) + sys.exit("Usage: {0} scope role datadir connstring".format(sys.argv[0])) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 2ee6e32a..1caa6229 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -122,7 +122,8 @@ class TestPostgresql(unittest.TestCase): 'callbacks': {'on_start': '/usr/bin/true', 'on_stop': '/usr/bin/true', 'on_restart': '/usr/bin/true', 'on_role_change': '/bin/true', 'on_reload': '/usr/bin/true' - }}) + }, + 'restore': '/usr/bin/true'}) psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) From 2089a123c2b1eb5dcc83b52234d00dc7afe86df1 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 3 Aug 2015 11:48:02 +0200 Subject: [PATCH 02/13] fix line too long warning. --- helpers/postgresql.py | 3 ++- scripts/restore.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index a7810573..7a92d42c 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -152,7 +152,8 @@ class Postgresql: def create_replica(self, master_connection, env): connstring = self.build_connstring(master_connection, master_connection) cmd = self.config['restore'] - ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", self.data_dir, connstring], env=env) + ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", + self.data_dir, connstring], env=env) return ret def is_leader(self, check_only=False): diff --git a/scripts/restore.py b/scripts/restore.py index f035c794..8dcb938c 100644 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -48,7 +48,8 @@ class Restore: def create_replica_with_pg_basebackup(self): master_connection = self.parse_connstring() ret = subprocess.call(['pg_basebackup', '-R', '-D', self.data_dir, '--host=' + master_connection['host'], - '--port=' + str(master_connection['port']), '-U', master_connection['user']], env=self.env) + '--port=' + str(master_connection['port']), '-U', master_connection['user']], + env=self.env) self.delete_trigger_file() return ret From fd884a53a94c69c46ff1fd2fd96955726f58bff0 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 3 Aug 2015 15:51:40 +0200 Subject: [PATCH 03/13] Multiple small fixes: - Do delete trigger file in the patroni itself. - Fix the typo in the datadir member. - Add #! at the top of the python script. --- helpers/postgresql.py | 1 + scripts/restore.py | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) mode change 100644 => 100755 scripts/restore.py diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 7a92d42c..7bb5b0ed 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -154,6 +154,7 @@ class Postgresql: cmd = self.config['restore'] ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", self.data_dir, connstring], env=env) + self.delete_trigger_file() return ret def is_leader(self, check_only=False): diff --git a/scripts/restore.py b/scripts/restore.py old mode 100644 new mode 100755 index 8dcb938c..3b95bc21 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -1,3 +1,4 @@ +#!/usr/bin/python # arguments are: # - cluster scope # - cluster role @@ -17,7 +18,7 @@ class Restore: self.scope = scope self.role = role self.connstring = connstring - self.datadir = datadir + self.data_dir = datadir self.env = os.environ.copy() def parse_connstring(self): @@ -50,7 +51,6 @@ class Restore: ret = subprocess.call(['pg_basebackup', '-R', '-D', self.data_dir, '--host=' + master_connection['host'], '--port=' + str(master_connection['port']), '-U', master_connection['user']], env=self.env) - self.delete_trigger_file() return ret From f17b651107256af532c0d9609a23cfab808952cd Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 4 Aug 2015 14:10:11 +0200 Subject: [PATCH 04/13] Adoprt WAL-E related code for a stand-alone script. - make sure all parameters are taken from the env variables. - make sure WAL-E code is not involved if the image is not configured for it - pick up pg_basebackup as an alternative if WAL-E fails. The WAL-E support will be switched on in subsequent commits. --- scripts/restore.py | 77 ++++++++++++++++++++++++++++++++++++---------- 1 file changed, 61 insertions(+), 16 deletions(-) diff --git a/scripts/restore.py b/scripts/restore.py index 3b95bc21..1e0af644 100755 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -3,6 +3,14 @@ # - 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 @@ -17,16 +25,17 @@ class Restore: def __init__(self, scope, role, datadir, connstring): self.scope = scope self.role = role - self.connstring = connstring + self.master_connection = Restore.parse_connstring(connstring) self.data_dir = datadir self.env = os.environ.copy() - def parse_connstring(self): + @staticmethod + def parse_connstring(self, connstring): # the connection string is in the form host= port= user= # return the dictionary with all components as separare keys result = {} - if self.connstring: - for x in self.connstring.split(): + if connstring: + for x in connstring.split(): if x and '=' in x: key, val = x.split('=') result[key.strip()] = val.strip() @@ -47,9 +56,10 @@ class Restore: return ret def create_replica_with_pg_basebackup(self): - master_connection = self.parse_connstring() - ret = subprocess.call(['pg_basebackup', '-R', '-D', self.data_dir, '--host=' + master_connection['host'], - '--port=' + str(master_connection['port']), '-U', master_connection['user']], + 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) return ret @@ -58,25 +68,60 @@ class WALERestore(Restore): def __init__(self, scope, role, datadir, connstring): super(WALERestore, self).__init__(scope, role, datadir, connstring) + # check the environment variables + self.init_error = False + 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(self): return self.create_replica_with_s3 - return self.create_replica_with_pg_basebackup + return 1 def replica_fallback_method(self): return self.create_replica_with_pg_basebackup - def should_use_s3_to_create_replica(self, master_connection): + def should_use_s3_to_create_replica(self): """ determine whether it makes sense to use S3 and not pg_basebackup """ - if not self.wal_e or not self.wal_e_paselfth: + if self.init_error: return False - threshold_megabytes = self.wal_e.get('threshold_megabytes', 10240) - threshold_backup_size_percentage = self.wal_e.get('threshold_backup_size_percentage', 30) + 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_path.split() + ['backup-list', '--detail', 'LATEST']) + 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 @@ -120,7 +165,7 @@ class WALERestore(Restore): 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(**master_connection) + 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,)) @@ -138,10 +183,10 @@ class WALERestore(Restore): (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100) def create_replica_with_s3(self): - if not self.wal_e or not self.wal_e_path: + if self.init_error: return 1 - ret = subprocess.call(self.wal_e_path + ' backup-fetch {} LATEST'.format(self.data_dir), shell=True) + ret = subprocess.call(self.wal_e.cmd + ' backup-fetch {} LATEST'.format(self.data_dir), env=self.env) self.restore_configuration_files() return ret From 67da1b3a786dfbc4b34913ffcc1086745ca31cd4 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 4 Aug 2015 15:26:12 +0200 Subject: [PATCH 05/13] Make parse_connstring staticmethod less 'selfish' --- scripts/restore.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/restore.py b/scripts/restore.py index 1e0af644..84f0e1f1 100755 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -30,7 +30,7 @@ class Restore: self.env = os.environ.copy() @staticmethod - def parse_connstring(self, connstring): + def parse_connstring(connstring): # the connection string is in the form host= port= user= # return the dictionary with all components as separare keys result = {} From e69f01536935c01755f38248ed79faf7f3ed3da7 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 4 Aug 2015 16:22:46 +0200 Subject: [PATCH 06/13] switch to using WAL-E by default. --- scripts/restore.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/restore.py b/scripts/restore.py index 84f0e1f1..a372223c 100755 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -194,6 +194,6 @@ class WALERestore(Restore): if __name__ == '__main__': if len(sys.argv) == 5: # scope, role, datadir, connstring - restore = Restore(*(sys.argv[1:])) + restore = WALERestore(*(sys.argv[1:])) sys.exit(restore.run()) sys.exit("Usage: {0} scope role datadir connstring".format(sys.argv[0])) From 8bed553d30e28349e6e4eee3dafc6241671f4ad6 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 4 Aug 2015 16:52:03 +0200 Subject: [PATCH 07/13] Use new-style classes. --- scripts/restore.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/restore.py b/scripts/restore.py index a372223c..837c8498 100755 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -20,7 +20,7 @@ import sys logger = logging.getLogger(__name__) -class Restore: +class Restore(object): def __init__(self, scope, role, datadir, connstring): self.scope = scope From 74a1daad090988e3d9bf740e17873dce1e64b656 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 4 Aug 2015 17:13:36 +0200 Subject: [PATCH 08/13] fix a typo in the named tuple definition. --- scripts/restore.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/restore.py b/scripts/restore.py index 837c8498..446ef134 100755 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -75,8 +75,8 @@ class WALERestore(Restore): 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 = 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') From 5d1d59a42bf7da5bd42ba4cd88a82864c1748fce Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 7 Aug 2015 15:48:06 +0200 Subject: [PATCH 09/13] Add mock-based unit tests for the restore module. - add unit tests for the restore module - additional dependencies in requirement - harden the code that calls restore callbacks - remove WAL-E related code from postgresql.py --- helpers/postgresql.py | 14 +++--- requirements-py2.txt | 2 + requirements-py3.txt | 1 + scripts/restore.py | 47 +++++++++++------ tests/test_restore.py | 114 ++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 155 insertions(+), 23 deletions(-) create mode 100644 tests/test_restore.py diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 7bb5b0ed..4616c325 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -61,10 +61,6 @@ class Postgresql: self.is_promoted = False self._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir] - self.wal_e = config.get('wal_e', None) - if self.wal_e: - self.wal_e_path = 'envdir {} wal-e --aws-instance-profile '.\ - format(self.wal_e.get('env_dir', '/home/postgres/etc/wal-e.d/env')) self.local_address = self.get_local_address() connect_address = config.get('connect_address', None) or self.local_address @@ -152,9 +148,13 @@ class Postgresql: def create_replica(self, master_connection, env): connstring = self.build_connstring(master_connection, master_connection) cmd = self.config['restore'] - ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", - self.data_dir, connstring], env=env) - self.delete_trigger_file() + try: + ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", + self.data_dir, connstring], env=env) + self.delete_trigger_file() + except Exception as e: + logger.error("Error when creating replica: {0}".format(e)) + return 1 return ret def is_leader(self, check_only=False): diff --git a/requirements-py2.txt b/requirements-py2.txt index 012d87a1..cb6203c5 100644 --- a/requirements-py2.txt +++ b/requirements-py2.txt @@ -1,6 +1,8 @@ boto dnspython +mock psycopg2 PyYAML requests +six >= 1.7 kazoo>=2.2.1 diff --git a/requirements-py3.txt b/requirements-py3.txt index bf481736..2c28b835 100644 --- a/requirements-py3.txt +++ b/requirements-py3.txt @@ -1,4 +1,5 @@ boto +mock dnspython3 psycopg2 PyYAML diff --git a/scripts/restore.py b/scripts/restore.py index 446ef134..2dca3775 100755 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -17,17 +17,21 @@ 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): + 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() + self.env = os.environ.copy() if not env else env @staticmethod def parse_connstring(connstring): @@ -41,6 +45,9 @@ class Restore(object): result[key.strip()] = val.strip() return result + def setup(self): + pass + def replica_method(self): return self.create_replica_with_pg_basebackup @@ -50,26 +57,31 @@ class Restore(object): def run(self): """ creates a new replica using either pg_basebackup or WAL-E """ method_fn = self.replica_method() - ret = method_fn() + 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): - 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) + 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): - super(WALERestore, self).__init__(scope, role, datadir, connstring) + 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: @@ -105,9 +117,9 @@ class WALERestore(Restore): self.init_error = True def replica_method(self): - if self.should_use_s3_to_create_replica(self): + if self.should_use_s3_to_create_replica(): return self.create_replica_with_s3 - return 1 + return None def replica_fallback_method(self): return self.create_replica_with_pg_basebackup @@ -185,9 +197,11 @@ class WALERestore(Restore): def create_replica_with_s3(self): if self.init_error: return 1 - - ret = subprocess.call(self.wal_e.cmd + ' backup-fetch {} LATEST'.format(self.data_dir), env=self.env) - self.restore_configuration_files() + 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 @@ -195,5 +209,6 @@ 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/tests/test_restore.py b/tests/test_restore.py new file mode 100644 index 00000000..cb4f2a83 --- /dev/null +++ b/tests/test_restore.py @@ -0,0 +1,114 @@ +import unittest +from mock import MagicMock, patch +import os +from 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) From 164b5760a4eee22621e931b67877c73f7d3dd1eb Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 7 Aug 2015 16:35:00 +0200 Subject: [PATCH 10/13] remove line too long warnings where possible. --- scripts/restore.py | 4 +++- tests/test_restore.py | 3 --- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/scripts/restore.py b/scripts/restore.py index 2dca3775..4ac091f3 100755 --- a/scripts/restore.py +++ b/scripts/restore.py @@ -74,6 +74,7 @@ class Restore(object): return 1 return ret + class WALERestore(Restore): def __init__(self, scope, role, datadir, connstring, env=None): @@ -133,7 +134,8 @@ class WALERestore(Restore): 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) + 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 diff --git a/tests/test_restore.py b/tests/test_restore.py index cb4f2a83..38b34c6a 100644 --- a/tests/test_restore.py +++ b/tests/test_restore.py @@ -50,7 +50,6 @@ class TestRestore(unittest.TestCase): 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)) @@ -76,7 +75,6 @@ class TestWALERestore(unittest.TestCase): 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) @@ -85,7 +83,6 @@ class TestWALERestore(unittest.TestCase): 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. From a07a80806aeec1341bb24fc6e447e776c6e50b5d Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 11 Aug 2015 11:33:40 +0200 Subject: [PATCH 11/13] Update postgresql.py Make build_connstring truly static --- helpers/postgresql.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index b9d13c16..b85bdac4 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -140,11 +140,11 @@ class Postgresql: return self.create_replica(r, env) == 0 @staticmethod - def build_connstring(self, conn): + def build_connstring(conn): return "host={host} port={port} user={user}".format(**conn) def create_replica(self, master_connection, env): - connstring = self.build_connstring(master_connection, master_connection) + connstring = self.build_connstring(master_connection) cmd = self.config['restore'] try: ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", From 72043703f5c97cb10b00b9efb4e0886544d23318 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 17 Aug 2015 16:06:48 +0200 Subject: [PATCH 12/13] Small code cleanup --- helpers/postgresql.py | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index b85bdac4..38f54904 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -145,13 +145,12 @@ class Postgresql: def create_replica(self, master_connection, env): connstring = self.build_connstring(master_connection) - cmd = self.config['restore'] + cmd = os.path.abspath(self.config['restore']) try: - ret = subprocess.call(shlex.split(os.path.abspath(cmd))+[self.scope, "replica", - self.data_dir, connstring], env=env) + ret = subprocess.call(shlex.split(cmd) + [self.scope, "replica", self.data_dir, connstring], env=env) self.delete_trigger_file() - except Exception as e: - logger.error("Error when creating replica: {0}".format(e)) + except: + logger.exception('Error when creating replica') return 1 return ret @@ -169,19 +168,18 @@ class Postgresql: """ pick a callback command and call it without waiting for it to finish """ if not self.callback or cb_name not in self.callback: return False - cmd = self.callback[cb_name] + cmd = os.path.abspath(self.callback[cb_name]) if is_leader is None: try: is_leader = self.is_leader(check_only=True) except psycopg2.OperationalError as e: logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e)) return False - scope = self.scope try: role = "master" if is_leader else "replica" - subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, role, scope]) - except Exception as e: - logger.warning("callback {0} {1} {2} {3} failed: {4}".format(os.path.abspath(cmd), cb_name, role, scope, e)) + subprocess.Popen(shlex.split(cmd) + [cb_name, role, self.scope]) + except: + logger.exception('callback %s %s %s %s failed', cmd, cb_name, role, self.scope) return False return True @@ -334,8 +332,8 @@ primary_conninfo = '{}' try: for f in self.configuration_to_save: shutil.copy(f + '.backup', f) - except Exception as e: - logger.error("unable to restore configuration from WAL-E backup: {}".format(e)) + except: + logger.exception('unable to restore configuration from WAL-E backup') def promote(self): self.is_promoted = subprocess.call(self._pg_ctl + ['promote']) == 0 From 29434e24ea79f81729d8e94913848d7e06b4639b Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 24 Aug 2015 16:53:57 +0200 Subject: [PATCH 13/13] Allow relative paths to external scripts. --- helpers/postgresql.py | 4 ++-- postgres0.yml | 2 +- tests/test_postgresql.py | 6 +++--- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 38f54904..ce8ca18f 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -145,7 +145,7 @@ class Postgresql: def create_replica(self, master_connection, env): connstring = self.build_connstring(master_connection) - cmd = os.path.abspath(self.config['restore']) + cmd = self.config['restore'] try: ret = subprocess.call(shlex.split(cmd) + [self.scope, "replica", self.data_dir, connstring], env=env) self.delete_trigger_file() @@ -168,7 +168,7 @@ class Postgresql: """ pick a callback command and call it without waiting for it to finish """ if not self.callback or cb_name not in self.callback: return False - cmd = os.path.abspath(self.callback[cb_name]) + cmd = self.callback[cb_name] if is_leader is None: try: is_leader = self.is_leader(check_only=True) diff --git a/postgres0.yml b/postgres0.yml index 200792d5..a2a7ce44 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -46,7 +46,7 @@ postgresql: env_dir: /home/postgres/etc/wal-e.d/env threshold_megabytes: 10240 threshold_backup_size_percentage: 30 - restore: /usr/bin/true + restore: "true" #recovery_conf: #restore_command: cp ../wal_archive/%f %p parameters: diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 6cbb6cb7..c4a3d6df 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -118,9 +118,9 @@ class TestPostgresql(unittest.TestCase): 'password': 'rep-pass', 'network': '127.0.0.1/32'}, 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}, - 'callbacks': {'on_start': '/usr/bin/true', 'on_stop': '/usr/bin/true', - 'on_restart': '/usr/bin/true', 'on_role_change': '/bin/true', - 'on_reload': '/usr/bin/true' + 'callbacks': {'on_start': 'true', 'on_stop': 'true', + 'on_restart': 'true', 'on_role_change': 'true', + 'on_reload': 'true' }, 'restore': '/usr/bin/true'}) psycopg2.connect = psycopg2_connect