# you need os for unittest to work
import os
import gc
from sys import exit, argv, version_info
import subprocess
import shutil
import six
import testgres
import hashlib
import re
import getpass
import select
from time import sleep
import re
import json
idx_ptrack = {
't_heap': {
'type': 'heap'
},
't_btree': {
'type': 'btree',
'column': 'text',
'relation': 't_heap'
},
't_seq': {
'type': 'seq',
'column': 't_seq',
'relation': 't_heap'
},
't_spgist': {
'type': 'spgist',
'column': 'text',
'relation': 't_heap'
},
't_brin': {
'type': 'brin',
'column': 'text',
'relation': 't_heap'
},
't_gist': {
'type': 'gist',
'column': 'tsvector',
'relation': 't_heap'
},
't_gin': {
'type': 'gin',
'column': 'tsvector',
'relation': 't_heap'
},
't_hash': {
'type': 'hash',
'column': 'id',
'relation': 't_heap'
},
't_bloom': {
'type': 'bloom',
'column': 'id',
'relation': 't_heap'
}
}
warning = """
Wrong splint in show_pb
Original Header:
{header}
Original Body:
{body}
Splitted Header
{header_split}
Splitted Body
{body_split}
"""
def dir_files(base_dir):
out_list = []
for dir_name, subdir_list, file_list in os.walk(base_dir):
if dir_name != base_dir:
out_list.append(os.path.relpath(dir_name, base_dir))
for fname in file_list:
out_list.append(
os.path.relpath(os.path.join(
dir_name, fname), base_dir)
)
out_list.sort()
return out_list
def is_enterprise():
# pg_config --help
cmd = [os.environ['PG_CONFIG'], '--help']
p = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
return b'postgrespro.ru' in p.communicate()[0]
def is_nls_enabled():
cmd = [os.environ['PG_CONFIG'], '--configure']
p = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
return b'enable-nls' in p.communicate()[0]
class ProbackupException(Exception):
def __init__(self, message, cmd):
self.message = message
self.cmd = cmd
def __str__(self):
return '\n ERROR: {0}\n CMD: {1}'.format(repr(self.message), self.cmd)
def slow_start(self, replica=False):
# wait for https://github.com/postgrespro/testgres/pull/50
# self.start()
# self.poll_query_until(
# "postgres",
# "SELECT not pg_is_in_recovery()",
# suppress={testgres.NodeConnection})
if replica:
query = 'SELECT pg_is_in_recovery()'
else:
query = 'SELECT not pg_is_in_recovery()'
self.start()
while True:
try:
output = self.safe_psql('template1', query).decode("utf-8").rstrip()
if output == 't':
break
except testgres.QueryException as e:
if 'database system is starting up' in e.message:
pass
elif 'FATAL: the database system is not accepting connections' in e.message:
pass
elif replica and 'Hot standby mode is disabled' in e.message:
raise e
else:
raise e
sleep(0.5)
class ProbackupTest(object):
# Class attributes
enterprise = is_enterprise()
enable_nls = is_nls_enabled()
def __init__(self, *args, **kwargs):
super(ProbackupTest, self).__init__(*args, **kwargs)
if '-v' in argv or '--verbose' in argv:
self.verbose = True
else:
self.verbose = False
self.test_env = os.environ.copy()
envs_list = [
'LANGUAGE',
'LC_ALL',
'PGCONNECT_TIMEOUT',
'PGDATA',
'PGDATABASE',
'PGHOSTADDR',
'PGREQUIRESSL',
'PGSERVICE',
'PGSSLMODE',
'PGUSER',
'PGPORT',
'PGHOST'
]
for e in envs_list:
try:
del self.test_env[e]
except:
pass
self.test_env['LC_MESSAGES'] = 'C'
self.test_env['LC_TIME'] = 'C'
self.gdb = 'PGPROBACKUP_GDB' in self.test_env and \
self.test_env['PGPROBACKUP_GDB'] == 'ON'
self.paranoia = 'PG_PROBACKUP_PARANOIA' in self.test_env and \
self.test_env['PG_PROBACKUP_PARANOIA'] == 'ON'
self.archive_compress = 'ARCHIVE_COMPRESSION' in self.test_env and \
self.test_env['ARCHIVE_COMPRESSION'] == 'ON'
try:
testgres.configure_testgres(
cache_initdb=False,
cached_initdb_dir=False,
cache_pg_config=False,
node_cleanup_full=False)
except:
pass
self.helpers_path = os.path.dirname(os.path.realpath(__file__))
self.dir_path = os.path.abspath(
os.path.join(self.helpers_path, os.pardir)
)
self.tmp_path = os.path.abspath(
os.path.join(self.dir_path, 'tmp_dirs')
)
try:
os.makedirs(os.path.join(self.dir_path, 'tmp_dirs'))
except:
pass
self.user = self.get_username()
self.probackup_path = None
if 'PGPROBACKUPBIN' in self.test_env:
if (
os.path.isfile(self.test_env["PGPROBACKUPBIN"]) and
os.access(self.test_env["PGPROBACKUPBIN"], os.X_OK)
):
self.probackup_path = self.test_env["PGPROBACKUPBIN"]
else:
if self.verbose:
print('PGPROBACKUPBIN is not an executable file')
if not self.probackup_path:
probackup_path_tmp = os.path.join(
testgres.get_pg_config()['BINDIR'], 'pg_probackup')
if os.path.isfile(probackup_path_tmp):
if not os.access(probackup_path_tmp, os.X_OK):
print('{0} is not an executable file'.format(
probackup_path_tmp))
else:
self.probackup_path = probackup_path_tmp
if not self.probackup_path:
probackup_path_tmp = os.path.abspath(os.path.join(
self.dir_path, '../pg_probackup'))
if os.path.isfile(probackup_path_tmp):
if not os.access(probackup_path_tmp, os.X_OK):
print('{0} is not an executable file'.format(
probackup_path_tmp))
else:
self.probackup_path = probackup_path_tmp
if not self.probackup_path:
print('pg_probackup binary is not found')
exit(1)
if os.name == 'posix':
self.EXTERNAL_DIRECTORY_DELIMITER = ':'
os.environ['PATH'] = os.path.dirname(
self.probackup_path) + ':' + os.environ['PATH']
elif os.name == 'nt':
self.EXTERNAL_DIRECTORY_DELIMITER = ';'
os.environ['PATH'] = os.path.dirname(
self.probackup_path) + ';' + os.environ['PATH']
self.probackup_old_path = None
if 'PGPROBACKUPBIN_OLD' in self.test_env:
if (
os.path.isfile(self.test_env['PGPROBACKUPBIN_OLD']) and
os.access(self.test_env['PGPROBACKUPBIN_OLD'], os.X_OK)
):
self.probackup_old_path = self.test_env['PGPROBACKUPBIN_OLD']
else:
if self.verbose:
print('PGPROBACKUPBIN_OLD is not an executable file')
self.probackup_version = None
self.old_probackup_version = None
try:
self.probackup_version_output = subprocess.check_output(
[self.probackup_path, "--version"],
stderr=subprocess.STDOUT,
).decode('utf-8')
except subprocess.CalledProcessError as e:
raise ProbackupException(e.output.decode('utf-8'))
if self.probackup_old_path:
old_probackup_version_output = subprocess.check_output(
[self.probackup_old_path, "--version"],
stderr=subprocess.STDOUT,
).decode('utf-8')
self.old_probackup_version = re.search(
r"\d+\.\d+\.\d+",
subprocess.check_output(
[self.probackup_old_path, "--version"],
stderr=subprocess.STDOUT,
).decode('utf-8')
).group(0)
self.probackup_version = re.search(r"\d+\.\d+\.\d+", self.probackup_version_output).group(0)
self.remote = False
self.remote_host = None
self.remote_port = None
self.remote_user = None
if 'PGPROBACKUP_SSH_REMOTE' in self.test_env:
if self.test_env['PGPROBACKUP_SSH_REMOTE'] == 'ON':
self.remote = True
self.ptrack = False
if 'PG_PROBACKUP_PTRACK' in self.test_env:
if self.test_env['PG_PROBACKUP_PTRACK'] == 'ON':
if self.pg_config_version >= self.version_to_num('11.0'):
self.ptrack = True
os.environ["PGAPPNAME"] = "pg_probackup"
@property
def pg_config_version(self):
return self.version_to_num(
testgres.get_pg_config()['VERSION'].split(" ")[1])
# if 'PGPROBACKUP_SSH_HOST' in self.test_env:
# self.remote_host = self.test_env['PGPROBACKUP_SSH_HOST']
# else
# print('PGPROBACKUP_SSH_HOST is not set')
# exit(1)
#
# if 'PGPROBACKUP_SSH_PORT' in self.test_env:
# self.remote_port = self.test_env['PGPROBACKUP_SSH_PORT']
# else
# print('PGPROBACKUP_SSH_PORT is not set')
# exit(1)
#
# if 'PGPROBACKUP_SSH_USER' in self.test_env:
# self.remote_user = self.test_env['PGPROBACKUP_SSH_USER']
# else
# print('PGPROBACKUP_SSH_USER is not set')
# exit(1)
def make_empty_node(
self,
base_dir=None):
real_base_dir = os.path.join(self.tmp_path, base_dir)
shutil.rmtree(real_base_dir, ignore_errors=True)
os.makedirs(real_base_dir)
node = testgres.get_new_node('test', base_dir=real_base_dir)
# bound method slow_start() to 'node' class instance
node.slow_start = slow_start.__get__(node)
node.should_rm_dirs = True
return node
def make_simple_node(
self,
base_dir=None,
set_replication=False,
ptrack_enable=False,
initdb_params=[],
pg_options={}):
node = self.make_empty_node(base_dir)
node.init(
initdb_params=initdb_params, allow_streaming=set_replication)
# set major version
with open(os.path.join(node.data_dir, 'PG_VERSION')) as f:
node.major_version_str = str(f.read().rstrip())
node.major_version = float(node.major_version_str)
# Sane default parameters
options = {}
options['max_connections'] = 100
options['shared_buffers'] = '10MB'
options['fsync'] = 'off'
options['wal_level'] = 'logical'
options['hot_standby'] = 'off'
options['log_line_prefix'] = '%t [%p]: [%l-1] '
options['log_statement'] = 'none'
options['log_duration'] = 'on'
options['log_min_duration_statement'] = 0
options['log_connections'] = 'on'
options['log_disconnections'] = 'on'
options['restart_after_crash'] = 'off'
options['autovacuum'] = 'off'
# Allow replication in pg_hba.conf
if set_replication:
options['max_wal_senders'] = 10
if ptrack_enable:
options['ptrack.map_size'] = '128'
options['shared_preload_libraries'] = 'ptrack'
if node.major_version >= 13:
options['wal_keep_size'] = '200MB'
else:
options['wal_keep_segments'] = '100'
# set default values
self.set_auto_conf(node, options)
# Apply given parameters
self.set_auto_conf(node, pg_options)
# kludge for testgres
# https://github.com/postgrespro/testgres/issues/54
# for PG >= 13 remove 'wal_keep_segments' parameter
if node.major_version >= 13:
self.set_auto_conf(
node, {}, 'postgresql.conf', ['wal_keep_segments'])
return node
def simple_bootstrap(self, node, role) -> None:
node.safe_psql(
'postgres',
'CREATE ROLE {0} WITH LOGIN REPLICATION'.format(role))
# PG 9.5
if self.get_version(node) < 90600:
node.safe_psql(
'postgres',
'GRANT USAGE ON SCHEMA pg_catalog TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.current_setting(text) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_is_in_recovery() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_start_backup(text, boolean) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_stop_backup() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_create_restore_point(text) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_switch_xlog() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_current() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_current_snapshot() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_snapshot_xmax(txid_snapshot) TO {0};'.format(role))
# PG 9.6
elif self.get_version(node) > 90600 and self.get_version(node) < 100000:
node.safe_psql(
'postgres',
'GRANT USAGE ON SCHEMA pg_catalog TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.current_setting(text) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_is_in_recovery() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_start_backup(text, boolean, boolean) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_stop_backup(boolean) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_create_restore_point(text) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_switch_xlog() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_last_xlog_replay_location() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_current() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_current_snapshot() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_snapshot_xmax(txid_snapshot) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_control_checkpoint() TO {0};'.format(role))
# >= 10
else:
node.safe_psql(
'postgres',
'GRANT USAGE ON SCHEMA pg_catalog TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.current_setting(text) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_is_in_recovery() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_start_backup(text, boolean, boolean) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_stop_backup(boolean, boolean) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_create_restore_point(text) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_switch_wal() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_last_wal_replay_lsn() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_current() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_current_snapshot() TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.txid_snapshot_xmax(txid_snapshot) TO {0}; '
'GRANT EXECUTE ON FUNCTION pg_catalog.pg_control_checkpoint() TO {0};'.format(role))
def create_tblspace_in_node(self, node, tblspc_name, tblspc_path=None, cfs=False):
res = node.execute(
'postgres',
'select exists'
" (select 1 from pg_tablespace where spcname = '{0}')".format(
tblspc_name)
)
# Check that tablespace with name 'tblspc_name' do not exists already
self.assertFalse(
res[0][0],
'Tablespace "{0}" already exists'.format(tblspc_name)
)
if not tblspc_path:
tblspc_path = os.path.join(
node.base_dir, '{0}'.format(tblspc_name))
cmd = "CREATE TABLESPACE {0} LOCATION '{1}'".format(
tblspc_name, tblspc_path)
if cfs:
cmd += ' with (compression=true)'
if not os.path.exists(tblspc_path):
os.makedirs(tblspc_path)
res = node.safe_psql('postgres', cmd)
# Check that tablespace was successfully created
# self.assertEqual(
# res[0], 0,
# 'Failed to create tablespace with cmd: {0}'.format(cmd))
def drop_tblspace(self, node, tblspc_name):
res = node.execute(
'postgres',
'select exists'
" (select 1 from pg_tablespace where spcname = '{0}')".format(
tblspc_name)
)
# Check that tablespace with name 'tblspc_name' do not exists already
self.assertTrue(
res[0][0],
'Tablespace "{0}" do not exists'.format(tblspc_name)
)
rels = node.execute(
"postgres",
"SELECT relname FROM pg_class c "
"LEFT JOIN pg_tablespace t ON c.reltablespace = t.oid "
"where c.relkind = 'r' and t.spcname = '{0}'".format(tblspc_name))
for rel in rels:
node.safe_psql(
'postgres',
"DROP TABLE {0}".format(rel[0]))
node.safe_psql(
'postgres',
'DROP TABLESPACE {0}'.format(tblspc_name))
def get_tblspace_path(self, node, tblspc_name):
return os.path.join(node.base_dir, tblspc_name)
def get_fork_size(self, node, fork_name):
return node.execute(
'postgres',
"select pg_relation_size('{0}')/8192".format(fork_name))[0][0]
def get_fork_path(self, node, fork_name):
return os.path.join(
node.base_dir, 'data', node.execute(
'postgres',
"select pg_relation_filepath('{0}')".format(
fork_name))[0][0]
)
def get_md5_per_page_for_fork(self, file, size_in_pages):
pages_per_segment = {}
md5_per_page = {}
size_in_pages = int(size_in_pages)
nsegments = int(size_in_pages/131072)
if size_in_pages % 131072 != 0:
nsegments = nsegments + 1
size = size_in_pages
for segment_number in range(nsegments):
if size - 131072 > 0:
pages_per_segment[segment_number] = 131072
else:
pages_per_segment[segment_number] = size
size = size - 131072
for segment_number in range(nsegments):
offset = 0
if segment_number == 0:
file_desc = os.open(file, os.O_RDONLY)
start_page = 0
end_page = pages_per_segment[segment_number]
else:
file_desc = os.open(
file+'.{0}'.format(segment_number), os.O_RDONLY
)
start_page = max(md5_per_page)+1
end_page = end_page + pages_per_segment[segment_number]
for page in range(start_page, end_page):
md5_per_page[page] = hashlib.md5(
os.read(file_desc, 8192)).hexdigest()
offset += 8192
os.lseek(file_desc, offset, 0)
os.close(file_desc)
return md5_per_page
def get_ptrack_bits_per_page_for_fork(self, node, file, size=[]):
if self.get_pgpro_edition(node) == 'enterprise':
if self.get_version(node) < self.version_to_num('10.0'):
header_size = 48
else:
header_size = 24
else:
header_size = 24
ptrack_bits_for_fork = []
# TODO: use macro instead of hard coded 8KB
page_body_size = 8192-header_size
# Check that if main fork file size is 0, it`s ok
# to not having a _ptrack fork
if os.path.getsize(file) == 0:
return ptrack_bits_for_fork
byte_size = os.path.getsize(file + '_ptrack')
npages = int(byte_size/8192)
if byte_size % 8192 != 0:
print('Ptrack page is not 8k aligned')
exit(1)
file = os.open(file + '_ptrack', os.O_RDONLY)
for page in range(npages):
offset = 8192*page+header_size
os.lseek(file, offset, 0)
lots_of_bytes = os.read(file, page_body_size)
byte_list = [
lots_of_bytes[i:i+1] for i in range(len(lots_of_bytes))
]
for byte in byte_list:
# byte_inverted = bin(int(byte, base=16))[2:][::-1]
# bits = (byte >> x) & 1 for x in range(7, -1, -1)
byte_inverted = bin(ord(byte))[2:].rjust(8, '0')[::-1]
for bit in byte_inverted:
# if len(ptrack_bits_for_fork) < size:
ptrack_bits_for_fork.append(int(bit))
os.close(file)
return ptrack_bits_for_fork
def check_ptrack_map_sanity(self, node, idx_ptrack):
success = True
for i in idx_ptrack:
# get new size of heap and indexes. size calculated in pages
idx_ptrack[i]['new_size'] = self.get_fork_size(node, i)
# update path to heap and index files in case they`ve changed
idx_ptrack[i]['path'] = self.get_fork_path(node, i)
# calculate new md5sums for pages
idx_ptrack[i]['new_pages'] = self.get_md5_per_page_for_fork(
idx_ptrack[i]['path'], idx_ptrack[i]['new_size'])
# get ptrack for every idx
idx_ptrack[i]['ptrack'] = self.get_ptrack_bits_per_page_for_fork(
node, idx_ptrack[i]['path'],
[idx_ptrack[i]['old_size'], idx_ptrack[i]['new_size']])
# compare pages and check ptrack sanity
if not self.check_ptrack_sanity(idx_ptrack[i]):
success = False
self.assertTrue(
success, 'Ptrack has failed to register changes in data files')
def check_ptrack_sanity(self, idx_dict):
success = True
if idx_dict['new_size'] > idx_dict['old_size']:
size = idx_dict['new_size']
else:
size = idx_dict['old_size']
for PageNum in range(size):
if PageNum not in idx_dict['old_pages']:
# Page was not present before, meaning that relation got bigger
# Ptrack should be equal to 1
if idx_dict['ptrack'][PageNum] != 1:
if self.verbose:
print(
'File: {0}\n Page Number {1} of type {2} was added,'
' but ptrack value is {3}. THIS IS BAD'.format(
idx_dict['path'],
PageNum, idx_dict['type'],
idx_dict['ptrack'][PageNum])
)
# print(idx_dict)
success = False
continue
if PageNum not in idx_dict['new_pages']:
# Page is not present now, meaning that relation got smaller
# Ptrack should be equal to 1,
# We are not freaking out about false positive stuff
if idx_dict['ptrack'][PageNum] != 1:
if self.verbose:
print(
'File: {0}\n Page Number {1} of type {2} was deleted,'
' but ptrack value is {3}. THIS IS BAD'.format(
idx_dict['path'],
PageNum, idx_dict['type'],
idx_dict['ptrack'][PageNum])
)
continue
# Ok, all pages in new_pages that do not have
# corresponding page in old_pages are been dealt with.
# We can now safely proceed to comparing old and new pages
if idx_dict['new_pages'][
PageNum] != idx_dict['old_pages'][PageNum]:
# Page has been changed,
# meaning that ptrack should be equal to 1
if idx_dict['ptrack'][PageNum] != 1:
if self.verbose:
print(
'File: {0}\n Page Number {1} of type {2} was changed,'
' but ptrack value is {3}. THIS IS BAD'.format(
idx_dict['path'],
PageNum, idx_dict['type'],
idx_dict['ptrack'][PageNum])
)
print(
' Old checksumm: {0}\n'
' New checksumm: {1}'.format(
idx_dict['old_pages'][PageNum],
idx_dict['new_pages'][PageNum])
)
if PageNum == 0 and idx_dict['type'] == 'spgist':
if self.verbose:
print(
'SPGIST is a special snowflake, so don`t '
'fret about losing ptrack for blknum 0'
)
continue
success = False
else:
# Page has not been changed,
# meaning that ptrack should be equal to 0
if idx_dict['ptrack'][PageNum] != 0:
if self.verbose:
print(
'File: {0}\n Page Number {1} of type {2} was not changed,'
' but ptrack value is {3}'.format(
idx_dict['path'],
PageNum, idx_dict['type'],
idx_dict['ptrack'][PageNum]
)
)
return success
# self.assertTrue(
# success, 'Ptrack has failed to register changes in data files'
# )
def get_backup_filelist(self, backup_dir, instance, backup_id):
filelist_path = os.path.join(
backup_dir, 'backups',
instance, backup_id, 'backup_content.control')
with open(filelist_path, 'r') as f:
filelist_raw = f.read()
filelist_splitted = filelist_raw.splitlines()
filelist = {}
for line in filelist_splitted:
line = json.loads(line)
filelist[line['path']] = line
return filelist
# return dict of files from filelist A,
# which are not exists in filelist_B
def get_backup_filelist_diff(self, filelist_A, filelist_B):
filelist_diff = {}
for file in filelist_A:
if file not in filelist_B:
filelist_diff[file] = filelist_A[file]
return filelist_diff
# used for partial restore
def truncate_every_file_in_dir(self, path):
for file in os.listdir(path):
with open(os.path.join(path, file), "w") as f:
f.close()
def check_ptrack_recovery(self, idx_dict):
size = idx_dict['size']
for PageNum in range(size):
if idx_dict['ptrack'][PageNum] != 1:
self.assertTrue(
False,
'Recovery for Page Number {0} of Type {1}'
' was conducted, but ptrack value is {2}.'
' THIS IS BAD\n IDX_DICT: {3}'.format(
PageNum, idx_dict['type'],
idx_dict['ptrack'][PageNum],
idx_dict
)
)
def check_ptrack_clean(self, idx_dict, size):
for PageNum in range(size):
if idx_dict['ptrack'][PageNum] != 0:
self.assertTrue(
False,
'Ptrack for Page Number {0} of Type {1}'
' should be clean, but ptrack value is {2}.'
'\n THIS IS BAD\n IDX_DICT: {3}'.format(
PageNum,
idx_dict['type'],
idx_dict['ptrack'][PageNum],
idx_dict
)
)
def run_pb(self, command, asynchronous=False, gdb=False, old_binary=False, return_id=True, env=None):
if not self.probackup_old_path and old_binary:
print('PGPROBACKUPBIN_OLD is not set')
exit(1)
if old_binary:
binary_path = self.probackup_old_path
else:
binary_path = self.probackup_path
if not env:
env=self.test_env
try:
self.cmd = [' '.join(map(str, [binary_path] + command))]
if self.verbose:
print(self.cmd)
if gdb:
return GDBobj([binary_path] + command, self)
if asynchronous:
return subprocess.Popen(
[binary_path] + command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env
)
else:
self.output = subprocess.check_output(
[binary_path] + command,
stderr=subprocess.STDOUT,
env=env
).decode('utf-8')
if command[0] == 'backup' and return_id:
# return backup ID
for line in self.output.splitlines():
if 'INFO: Backup' and 'completed' in line:
return line.split()[2]
else:
return self.output
except subprocess.CalledProcessError as e:
raise ProbackupException(e.output.decode('utf-8'), self.cmd)
def run_binary(self, command, asynchronous=False, env=None):
if not env:
env = self.test_env
if self.verbose:
print([' '.join(map(str, command))])
try:
if asynchronous:
return subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env
)
else:
self.output = subprocess.check_output(
command,
stderr=subprocess.STDOUT,
env=env
).decode('utf-8')
return self.output
except subprocess.CalledProcessError as e:
raise ProbackupException(e.output.decode('utf-8'), command)
def init_pb(self, backup_dir, options=[], old_binary=False):
shutil.rmtree(backup_dir, ignore_errors=True)
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
return self.run_pb([
'init',
'-B', backup_dir
] + options,
old_binary=old_binary
)
def add_instance(self, backup_dir, instance, node, old_binary=False, options=[]):
cmd = [
'add-instance',
'--instance={0}'.format(instance),
'-B', backup_dir,
'-D', node.data_dir
]
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
return self.run_pb(cmd + options, old_binary=old_binary)
def set_config(self, backup_dir, instance, old_binary=False, options=[]):
cmd = [
'set-config',
'--instance={0}'.format(instance),
'-B', backup_dir,
]
return self.run_pb(cmd + options, old_binary=old_binary)
def set_backup(self, backup_dir, instance, backup_id=False,
old_binary=False, options=[]):
cmd = [
'set-backup',
'-B', backup_dir
]
if instance:
cmd = cmd + ['--instance={0}'.format(instance)]
if backup_id:
cmd = cmd + ['-i', backup_id]
return self.run_pb(cmd + options, old_binary=old_binary)
def del_instance(self, backup_dir, instance, old_binary=False):
return self.run_pb([
'del-instance',
'--instance={0}'.format(instance),
'-B', backup_dir
],
old_binary=old_binary
)
def clean_pb(self, backup_dir):
shutil.rmtree(backup_dir, ignore_errors=True)
def backup_node(
self, backup_dir, instance, node, data_dir=False,
backup_type='full', datname=False, options=[],
asynchronous=False, gdb=False,
old_binary=False, return_id=True, no_remote=False,
env=None
):
if not node and not data_dir:
print('You must provide ether node or data_dir for backup')
exit(1)
if not datname:
datname = 'postgres'
cmd_list = [
'backup',
'-B', backup_dir,
'--instance={0}'.format(instance),
# "-D", pgdata,
'-p', '%i' % node.port,
'-d', datname
]
if data_dir:
cmd_list += ['-D', data_dir]
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary and not no_remote:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
if backup_type:
cmd_list += ['-b', backup_type]
if not old_binary:
cmd_list += ['--no-sync']
return self.run_pb(cmd_list + options, asynchronous, gdb, old_binary, return_id, env=env)
def checkdb_node(
self, backup_dir=False, instance=False, data_dir=False,
options=[], asynchronous=False, gdb=False, old_binary=False
):
cmd_list = ["checkdb"]
if backup_dir:
cmd_list += ["-B", backup_dir]
if instance:
cmd_list += ["--instance={0}".format(instance)]
if data_dir:
cmd_list += ["-D", data_dir]
return self.run_pb(cmd_list + options, asynchronous, gdb, old_binary)
def merge_backup(
self, backup_dir, instance, backup_id, asynchronous=False,
gdb=False, old_binary=False, options=[]):
cmd_list = [
'merge',
'-B', backup_dir,
'--instance={0}'.format(instance),
'-i', backup_id
]
return self.run_pb(cmd_list + options, asynchronous, gdb, old_binary)
def restore_node(
self, backup_dir, instance, node=False,
data_dir=None, backup_id=None, old_binary=False, options=[],
gdb=False
):
if data_dir is None:
data_dir = node.data_dir
cmd_list = [
'restore',
'-B', backup_dir,
'-D', data_dir,
'--instance={0}'.format(instance)
]
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
if backup_id:
cmd_list += ['-i', backup_id]
if not old_binary:
cmd_list += ['--no-sync']
return self.run_pb(cmd_list + options, gdb=gdb, old_binary=old_binary)
def catchup_node(
self,
backup_mode, source_pgdata, destination_node,
options = []
):
cmd_list = [
'catchup',
'--backup-mode={0}'.format(backup_mode),
'--source-pgdata={0}'.format(source_pgdata),
'--destination-pgdata={0}'.format(destination_node.data_dir)
]
if self.remote:
cmd_list += ['--remote-proto=ssh', '--remote-host=localhost']
if self.verbose:
cmd_list += [
'--log-level-file=VERBOSE',
'--log-directory={0}'.format(destination_node.logs_dir)
]
return self.run_pb(cmd_list + options)
def show_pb(
self, backup_dir, instance=None, backup_id=None,
options=[], as_text=False, as_json=True, old_binary=False,
env=None
):
backup_list = []
specific_record = {}
cmd_list = [
'show',
'-B', backup_dir,
]
if instance:
cmd_list += ['--instance={0}'.format(instance)]
if backup_id:
cmd_list += ['-i', backup_id]
# AHTUNG, WARNING will break json parsing
if as_json:
cmd_list += ['--format=json', '--log-level-console=error']
if as_text:
# You should print it when calling as_text=true
return self.run_pb(cmd_list + options, old_binary=old_binary, env=env)
# get show result as list of lines
if as_json:
data = json.loads(self.run_pb(cmd_list + options, old_binary=old_binary))
# print(data)
for instance_data in data:
# find specific instance if requested
if instance and instance_data['instance'] != instance:
continue
for backup in reversed(instance_data['backups']):
# find specific backup if requested
if backup_id:
if backup['id'] == backup_id:
return backup
else:
backup_list.append(backup)
if backup_id is not None:
self.assertTrue(False, "Failed to find backup with ID: {0}".format(backup_id))
return backup_list
else:
show_splitted = self.run_pb(
cmd_list + options, old_binary=old_binary, env=env).splitlines()
if instance is not None and backup_id is None:
# cut header(ID, Mode, etc) from show as single string
header = show_splitted[1:2][0]
# cut backup records from show as single list
# with string for every backup record
body = show_splitted[3:]
# inverse list so oldest record come first
body = body[::-1]
# split string in list with string for every header element
header_split = re.split(' +', header)
# Remove empty items
for i in header_split:
if i == '':
header_split.remove(i)
continue
header_split = [
header_element.rstrip() for header_element in header_split
]
for backup_record in body:
backup_record = backup_record.rstrip()
# split list with str for every backup record element
backup_record_split = re.split(' +', backup_record)
# Remove empty items
for i in backup_record_split:
if i == '':
backup_record_split.remove(i)
if len(header_split) != len(backup_record_split):
print(warning.format(
header=header, body=body,
header_split=header_split,
body_split=backup_record_split)
)
exit(1)
new_dict = dict(zip(header_split, backup_record_split))
backup_list.append(new_dict)
return backup_list
else:
# cut out empty lines and lines started with #
# and other garbage then reconstruct it as dictionary
# print show_splitted
sanitized_show = [item for item in show_splitted if item]
sanitized_show = [
item for item in sanitized_show if not item.startswith('#')
]
# print sanitized_show
for line in sanitized_show:
name, var = line.partition(' = ')[::2]
var = var.strip('"')
var = var.strip("'")
specific_record[name.strip()] = var
if not specific_record:
self.assertTrue(False, "Failed to find backup with ID: {0}".format(backup_id))
return specific_record
def show_archive(
self, backup_dir, instance=None, options=[],
as_text=False, as_json=True, old_binary=False,
tli=0
):
cmd_list = [
'show',
'--archive',
'-B', backup_dir,
]
if instance:
cmd_list += ['--instance={0}'.format(instance)]
# AHTUNG, WARNING will break json parsing
if as_json:
cmd_list += ['--format=json', '--log-level-console=error']
if as_text:
# You should print it when calling as_text=true
return self.run_pb(cmd_list + options, old_binary=old_binary)
if as_json:
if as_text:
data = self.run_pb(cmd_list + options, old_binary=old_binary)
else:
data = json.loads(self.run_pb(cmd_list + options, old_binary=old_binary))
if instance:
instance_timelines = None
for instance_name in data:
if instance_name['instance'] == instance:
instance_timelines = instance_name['timelines']
break
if tli > 0:
timeline_data = None
for timeline in instance_timelines:
if timeline['tli'] == tli:
return timeline
return {}
if instance_timelines:
return instance_timelines
return data
else:
show_splitted = self.run_pb(
cmd_list + options, old_binary=old_binary).splitlines()
print(show_splitted)
exit(1)
def validate_pb(
self, backup_dir, instance=None, backup_id=None,
options=[], old_binary=False, gdb=False, asynchronous=False
):
cmd_list = [
'validate',
'-B', backup_dir
]
if instance:
cmd_list += ['--instance={0}'.format(instance)]
if backup_id:
cmd_list += ['-i', backup_id]
return self.run_pb(cmd_list + options, old_binary=old_binary, gdb=gdb, asynchronous=asynchronous)
def delete_pb(
self, backup_dir, instance, backup_id=None,
options=[], old_binary=False, gdb=False, asynchronous=False):
cmd_list = [
'delete',
'-B', backup_dir
]
cmd_list += ['--instance={0}'.format(instance)]
if backup_id:
cmd_list += ['-i', backup_id]
return self.run_pb(cmd_list + options, old_binary=old_binary, gdb=gdb, asynchronous=asynchronous)
def delete_expired(
self, backup_dir, instance, options=[], old_binary=False):
cmd_list = [
'delete',
'-B', backup_dir,
'--instance={0}'.format(instance)
]
return self.run_pb(cmd_list + options, old_binary=old_binary)
def show_config(self, backup_dir, instance, old_binary=False):
out_dict = {}
cmd_list = [
'show-config',
'-B', backup_dir,
'--instance={0}'.format(instance)
]
res = self.run_pb(cmd_list, old_binary=old_binary).splitlines()
for line in res:
if not line.startswith('#'):
name, var = line.partition(' = ')[::2]
out_dict[name] = var
return out_dict
def get_recovery_conf(self, node):
out_dict = {}
if self.get_version(node) >= self.version_to_num('12.0'):
recovery_conf_path = os.path.join(node.data_dir, 'postgresql.auto.conf')
with open(recovery_conf_path, 'r') as f:
print(f.read())
else:
recovery_conf_path = os.path.join(node.data_dir, 'recovery.conf')
with open(
recovery_conf_path, 'r'
) as recovery_conf:
for line in recovery_conf:
try:
key, value = line.split('=')
except:
continue
out_dict[key.strip()] = value.strip(" '").replace("'\n", "")
return out_dict
def set_archiving(
self, backup_dir, instance, node, replica=False,
overwrite=False, compress=True, old_binary=False,
log_level=False, archive_timeout=False,
custom_archive_command=None):
# parse postgresql.auto.conf
options = {}
if replica:
options['archive_mode'] = 'always'
options['hot_standby'] = 'on'
else:
options['archive_mode'] = 'on'
if custom_archive_command is None:
if os.name == 'posix':
options['archive_command'] = '"{0}" archive-push -B {1} --instance={2} '.format(
self.probackup_path, backup_dir, instance)
elif os.name == 'nt':
options['archive_command'] = '"{0}" archive-push -B {1} --instance={2} '.format(
self.probackup_path.replace("\\","\\\\"),
backup_dir.replace("\\","\\\\"), instance)
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options['archive_command'] += '--remote-proto=ssh '
options['archive_command'] += '--remote-host=localhost '
if self.archive_compress and compress:
options['archive_command'] += '--compress '
if overwrite:
options['archive_command'] += '--overwrite '
options['archive_command'] += '--log-level-console=VERBOSE '
options['archive_command'] += '-j 5 '
options['archive_command'] += '--batch-size 10 '
options['archive_command'] += '--no-sync '
if archive_timeout:
options['archive_command'] += '--archive-timeout={0} '.format(
archive_timeout)
if os.name == 'posix':
options['archive_command'] += '--wal-file-path=%p --wal-file-name=%f'
elif os.name == 'nt':
options['archive_command'] += '--wal-file-path="%p" --wal-file-name="%f"'
if log_level:
options['archive_command'] += ' --log-level-console={0}'.format(log_level)
options['archive_command'] += ' --log-level-file={0} '.format(log_level)
else: # custom_archive_command is not None
options['archive_command'] = custom_archive_command
self.set_auto_conf(node, options)
def get_restore_command(self, backup_dir, instance, node):
# parse postgresql.auto.conf
restore_command = ''
if os.name == 'posix':
restore_command += '{0} archive-get -B {1} --instance={2} '.format(
self.probackup_path, backup_dir, instance)
elif os.name == 'nt':
restore_command += '"{0}" archive-get -B {1} --instance={2} '.format(
self.probackup_path.replace("\\","\\\\"),
backup_dir.replace("\\","\\\\"), instance)
# don`t forget to kill old_binary after remote ssh release
if self.remote:
restore_command += '--remote-proto=ssh '
restore_command += '--remote-host=localhost '
if os.name == 'posix':
restore_command += '--wal-file-path=%p --wal-file-name=%f'
elif os.name == 'nt':
restore_command += '--wal-file-path="%p" --wal-file-name="%f"'
return restore_command
# rm_options - list of parameter name that should be deleted from current config,
# example: ['wal_keep_segments', 'max_wal_size']
def set_auto_conf(self, node, options, config='postgresql.auto.conf', rm_options={}):
# parse postgresql.auto.conf
path = os.path.join(node.data_dir, config)
with open(path, 'r') as f:
raw_content = f.read()
current_options = {}
current_directives = []
for line in raw_content.splitlines():
# ignore comments
if line.startswith('#'):
continue
if line == '':
continue
if line.startswith('include'):
current_directives.append(line)
continue
name, var = line.partition('=')[::2]
name = name.strip()
var = var.strip()
var = var.strip('"')
var = var.strip("'")
# remove options specified in rm_options list
if name in rm_options:
continue
current_options[name] = var
for option in options:
current_options[option] = options[option]
auto_conf = ''
for option in current_options:
auto_conf += "{0} = '{1}'\n".format(
option, current_options[option])
for directive in current_directives:
auto_conf += directive + "\n"
with open(path, 'wt') as f:
f.write(auto_conf)
f.flush()
f.close()
def set_replica(
self, master, replica,
replica_name='replica',
synchronous=False,
log_shipping=False
):
self.set_auto_conf(
replica,
options={
'port': replica.port,
'hot_standby': 'on'})
if self.get_version(replica) >= self.version_to_num('12.0'):
with open(os.path.join(replica.data_dir, "standby.signal"), 'w') as f:
f.flush()
f.close()
config = 'postgresql.auto.conf'
if not log_shipping:
self.set_auto_conf(
replica,
{'primary_conninfo': 'user={0} port={1} application_name={2} '
' sslmode=prefer sslcompression=1'.format(
self.user, master.port, replica_name)},
config)
else:
replica.append_conf('recovery.conf', 'standby_mode = on')
if not log_shipping:
replica.append_conf(
'recovery.conf',
"primary_conninfo = 'user={0} port={1} application_name={2}"
" sslmode=prefer sslcompression=1'".format(
self.user, master.port, replica_name))
if synchronous:
self.set_auto_conf(
master,
options={
'synchronous_standby_names': replica_name,
'synchronous_commit': 'remote_apply'})
master.reload()
def change_backup_status(self, backup_dir, instance, backup_id, status):
control_file_path = os.path.join(
backup_dir, 'backups', instance, backup_id, 'backup.control')
with open(control_file_path, 'r') as f:
actual_control = f.read()
new_control_file = ''
for line in actual_control.splitlines():
if line.startswith('status'):
line = 'status = {0}'.format(status)
new_control_file += line
new_control_file += '\n'
with open(control_file_path, 'wt') as f:
f.write(new_control_file)
f.flush()
f.close()
with open(control_file_path, 'r') as f:
actual_control = f.read()
def wrong_wal_clean(self, node, wal_size):
wals_dir = os.path.join(self.backup_dir(node), 'wal')
wals = [
f for f in os.listdir(wals_dir) if os.path.isfile(
os.path.join(wals_dir, f))
]
wals.sort()
file_path = os.path.join(wals_dir, wals[-1])
if os.path.getsize(file_path) != wal_size:
os.remove(file_path)
def guc_wal_segment_size(self, node):
var = node.execute(
'postgres',
"select setting from pg_settings where name = 'wal_segment_size'"
)
return int(var[0][0]) * self.guc_wal_block_size(node)
def guc_wal_block_size(self, node):
var = node.execute(
'postgres',
"select setting from pg_settings where name = 'wal_block_size'"
)
return int(var[0][0])
def get_pgpro_edition(self, node):
if node.execute(
'postgres',
"select exists (select 1 from"
" pg_proc where proname = 'pgpro_edition')"
)[0][0]:
var = node.execute('postgres', 'select pgpro_edition()')
return str(var[0][0])
else:
return False
def get_username(self):
""" Returns current user name """
return getpass.getuser()
def version_to_num(self, version):
if not version:
return 0
parts = version.split('.')
while len(parts) < 3:
parts.append('0')
num = 0
for part in parts:
num = num * 100 + int(re.sub("[^\d]", "", part))
return num
def switch_wal_segment(self, node):
"""
Execute pg_switch_wal/xlog() in given node
Args:
node: an instance of PostgresNode or NodeConnection class
"""
if isinstance(node, testgres.PostgresNode):
if self.version_to_num(
node.safe_psql('postgres', 'show server_version').decode('utf-8')
) >= self.version_to_num('10.0'):
node.safe_psql('postgres', 'select pg_switch_wal()')
else:
node.safe_psql('postgres', 'select pg_switch_xlog()')
else:
if self.version_to_num(
node.execute('show server_version')[0][0]
) >= self.version_to_num('10.0'):
node.execute('select pg_switch_wal()')
else:
node.execute('select pg_switch_xlog()')
sleep(1)
def wait_until_replica_catch_with_master(self, master, replica):
version = master.safe_psql(
'postgres',
'show server_version').decode('utf-8').rstrip()
if self.version_to_num(version) >= self.version_to_num('10.0'):
master_function = 'pg_catalog.pg_current_wal_lsn()'
replica_function = 'pg_catalog.pg_last_wal_replay_lsn()'
else:
master_function = 'pg_catalog.pg_current_xlog_location()'
replica_function = 'pg_catalog.pg_last_xlog_replay_location()'
lsn = master.safe_psql(
'postgres',
'SELECT {0}'.format(master_function)).decode('utf-8').rstrip()
# Wait until replica catch up with master
replica.poll_query_until(
'postgres',
"SELECT '{0}'::pg_lsn <= {1}".format(lsn, replica_function))
def get_version(self, node):
return self.version_to_num(
testgres.get_pg_config()['VERSION'].split(" ")[1])
def get_ptrack_version(self, node):
version = node.safe_psql(
"postgres",
"SELECT extversion "
"FROM pg_catalog.pg_extension WHERE extname = 'ptrack'").decode('utf-8').rstrip()
return self.version_to_num(version)
def get_bin_path(self, binary):
return testgres.get_bin_path(binary)
def clean_all(self):
for o in gc.get_referrers(testgres.PostgresNode):
if o.__class__ is testgres.PostgresNode:
o.cleanup()
def del_test_dir(self, module_name, fname):
""" Del testdir and optimistically try to del module dir"""
self.clean_all()
shutil.rmtree(
os.path.join(
self.tmp_path,
module_name,
fname
),
ignore_errors=True
)
def pgdata_content(self, pgdata, ignore_ptrack=True, exclude_dirs=None):
""" return dict with directory content. "
" TAKE IT AFTER CHECKPOINT or BACKUP"""
dirs_to_ignore = [
'pg_xlog', 'pg_wal', 'pg_log',
'pg_stat_tmp', 'pg_subtrans', 'pg_notify'
]
files_to_ignore = [
'postmaster.pid', 'postmaster.opts',
'pg_internal.init', 'postgresql.auto.conf',
'backup_label', 'tablespace_map', 'recovery.conf',
'ptrack_control', 'ptrack_init', 'pg_control',
'probackup_recovery.conf', 'recovery.signal',
'standby.signal', 'ptrack.map', 'ptrack.map.mmap',
'ptrack.map.tmp'
]
if exclude_dirs:
dirs_to_ignore = dirs_to_ignore + exclude_dirs
# suffixes_to_ignore = (
# '_ptrack'
# )
directory_dict = {}
directory_dict['pgdata'] = pgdata
directory_dict['files'] = {}
directory_dict['dirs'] = {}
for root, dirs, files in os.walk(pgdata, followlinks=True):
dirs[:] = [d for d in dirs if d not in dirs_to_ignore]
for file in files:
if (
file in files_to_ignore or
(ignore_ptrack and file.endswith('_ptrack'))
):
continue
file_fullpath = os.path.join(root, file)
file_relpath = os.path.relpath(file_fullpath, pgdata)
directory_dict['files'][file_relpath] = {'is_datafile': False}
with open(file_fullpath, 'rb') as f:
directory_dict['files'][file_relpath]['md5'] = hashlib.md5(f.read()).hexdigest()
f.close()
# directory_dict['files'][file_relpath]['md5'] = hashlib.md5(
# f = open(file_fullpath, 'rb').read()).hexdigest()
# crappy algorithm
if file.isdigit():
directory_dict['files'][file_relpath]['is_datafile'] = True
size_in_pages = os.path.getsize(file_fullpath)/8192
directory_dict['files'][file_relpath][
'md5_per_page'] = self.get_md5_per_page_for_fork(
file_fullpath, size_in_pages
)
for root, dirs, files in os.walk(pgdata, topdown=False, followlinks=True):
for directory in dirs:
directory_path = os.path.join(root, directory)
directory_relpath = os.path.relpath(directory_path, pgdata)
found = False
for d in dirs_to_ignore:
if d in directory_relpath:
found = True
break
# check if directory already here as part of larger directory
if not found:
for d in directory_dict['dirs']:
# print("OLD dir {0}".format(d))
if directory_relpath in d:
found = True
break
if not found:
directory_dict['dirs'][directory_relpath] = {}
# get permissions for every file and directory
for file in directory_dict['dirs']:
full_path = os.path.join(pgdata, file)
directory_dict['dirs'][file]['mode'] = os.stat(
full_path).st_mode
for file in directory_dict['files']:
full_path = os.path.join(pgdata, file)
directory_dict['files'][file]['mode'] = os.stat(
full_path).st_mode
return directory_dict
def get_known_bugs_comparision_exclusion_dict(self, node):
""" get dict of known datafiles difference, that can be used in compare_pgdata() """
comparision_exclusion_dict = dict()
# bug in spgist metapage update (PGPRO-5707)
spgist_filelist = node.safe_psql(
"postgres",
"SELECT pg_catalog.pg_relation_filepath(pg_class.oid) "
"FROM pg_am, pg_class "
"WHERE pg_am.amname = 'spgist' "
"AND pg_class.relam = pg_am.oid"
).decode('utf-8').rstrip().splitlines()
for filename in spgist_filelist:
comparision_exclusion_dict[filename] = set([0])
return comparision_exclusion_dict
def compare_pgdata(self, original_pgdata, restored_pgdata, exclusion_dict = dict()):
"""
return dict with directory content. DO IT BEFORE RECOVERY
exclusion_dict is used for exclude files (and it block_no) from comparision
it is a dict with relative filenames as keys and set of block numbers as values
"""
fail = False
error_message = 'Restored PGDATA is not equal to original!\n'
# Compare directories
for directory in restored_pgdata['dirs']:
if directory not in original_pgdata['dirs']:
fail = True
error_message += '\nDirectory was not present'
error_message += ' in original PGDATA: {0}\n'.format(
os.path.join(restored_pgdata['pgdata'], directory))
else:
if (
restored_pgdata['dirs'][directory]['mode'] !=
original_pgdata['dirs'][directory]['mode']
):
fail = True
error_message += '\nDir permissions mismatch:\n'
error_message += ' Dir old: {0} Permissions: {1}\n'.format(
os.path.join(original_pgdata['pgdata'], directory),
original_pgdata['dirs'][directory]['mode'])
error_message += ' Dir new: {0} Permissions: {1}\n'.format(
os.path.join(restored_pgdata['pgdata'], directory),
restored_pgdata['dirs'][directory]['mode'])
for directory in original_pgdata['dirs']:
if directory not in restored_pgdata['dirs']:
fail = True
error_message += '\nDirectory dissappeared'
error_message += ' in restored PGDATA: {0}\n'.format(
os.path.join(restored_pgdata['pgdata'], directory))
for file in restored_pgdata['files']:
# File is present in RESTORED PGDATA
# but not present in ORIGINAL
# only backup_label is allowed
if file not in original_pgdata['files']:
fail = True
error_message += '\nFile is not present'
error_message += ' in original PGDATA: {0}\n'.format(
os.path.join(restored_pgdata['pgdata'], file))
for file in original_pgdata['files']:
if file in restored_pgdata['files']:
if (
restored_pgdata['files'][file]['mode'] !=
original_pgdata['files'][file]['mode']
):
fail = True
error_message += '\nFile permissions mismatch:\n'
error_message += ' File_old: {0} Permissions: {1:o}\n'.format(
os.path.join(original_pgdata['pgdata'], file),
original_pgdata['files'][file]['mode'])
error_message += ' File_new: {0} Permissions: {1:o}\n'.format(
os.path.join(restored_pgdata['pgdata'], file),
restored_pgdata['files'][file]['mode'])
if (
original_pgdata['files'][file]['md5'] !=
restored_pgdata['files'][file]['md5']
):
if file not in exclusion_dict:
fail = True
error_message += (
'\nFile Checksum mismatch.\n'
'File_old: {0}\nChecksum_old: {1}\n'
'File_new: {2}\nChecksum_new: {3}\n').format(
os.path.join(original_pgdata['pgdata'], file),
original_pgdata['files'][file]['md5'],
os.path.join(restored_pgdata['pgdata'], file),
restored_pgdata['files'][file]['md5']
)
if original_pgdata['files'][file]['is_datafile']:
for page in original_pgdata['files'][file]['md5_per_page']:
if page not in restored_pgdata['files'][file]['md5_per_page']:
error_message += (
'\n Page {0} dissappeared.\n '
'File: {1}\n').format(
page,
os.path.join(
restored_pgdata['pgdata'],
file
)
)
continue
if not (file in exclusion_dict and page in exclusion_dict[file]):
if (
original_pgdata['files'][file]['md5_per_page'][page] !=
restored_pgdata['files'][file]['md5_per_page'][page]
):
fail = True
error_message += (
'\n Page checksum mismatch: {0}\n '
' PAGE Checksum_old: {1}\n '
' PAGE Checksum_new: {2}\n '
' File: {3}\n'
).format(
page,
original_pgdata['files'][file][
'md5_per_page'][page],
restored_pgdata['files'][file][
'md5_per_page'][page],
os.path.join(
restored_pgdata['pgdata'], file)
)
for page in restored_pgdata['files'][file]['md5_per_page']:
if page not in original_pgdata['files'][file]['md5_per_page']:
error_message += '\n Extra page {0}\n File: {1}\n'.format(
page,
os.path.join(
restored_pgdata['pgdata'], file))
else:
error_message += (
'\nFile disappearance.\n '
'File: {0}\n').format(
os.path.join(restored_pgdata['pgdata'], file)
)
fail = True
self.assertFalse(fail, error_message)
def gdb_attach(self, pid):
return GDBobj([str(pid)], self, attach=True)
def _check_gdb_flag_or_skip_test(self):
if not self.gdb:
self.skipTest(
"Specify PGPROBACKUP_GDB and build without "
"optimizations for run this test"
)
class GdbException(Exception):
def __init__(self, message="False"):
self.message = message
def __str__(self):
return '\n ERROR: {0}\n'.format(repr(self.message))
class GDBobj:
def __init__(self, cmd, env, attach=False):
self.verbose = env.verbose
self.output = ''
# Check gdb flag is set up
if not env.gdb:
raise GdbException("No `PGPROBACKUP_GDB=on` is set, "
"test should call ProbackupTest::check_gdb_flag_or_skip_test() on its start "
"and be skipped")
# Check gdb presense
try:
gdb_version, _ = subprocess.Popen(
['gdb', '--version'],
stdout=subprocess.PIPE
).communicate()
except OSError:
raise GdbException("Couldn't find gdb on the path")
self.base_cmd = [
'gdb',
'--interpreter',
'mi2',
]
if attach:
self.cmd = self.base_cmd + ['--pid'] + cmd
else:
self.cmd = self.base_cmd + ['--args'] + cmd
# Get version
gdb_version_number = re.search(
b"^GNU gdb [^\d]*(\d+)\.(\d)",
gdb_version)
self.major_version = int(gdb_version_number.group(1))
self.minor_version = int(gdb_version_number.group(2))
if self.verbose:
print([' '.join(map(str, self.cmd))])
self.proc = subprocess.Popen(
self.cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=0,
universal_newlines=True
)
self.gdb_pid = self.proc.pid
while True:
line = self.get_line()
if 'No such process' in line:
raise GdbException(line)
if not line.startswith('(gdb)'):
pass
else:
break
def get_line(self):
line = self.proc.stdout.readline()
self.output += line
return line
def kill(self):
self.proc.kill()
self.proc.wait()
def set_breakpoint(self, location):
result = self._execute('break ' + location)
for line in result:
if line.startswith('~"Breakpoint'):
return
elif line.startswith('=breakpoint-created'):
return
elif line.startswith('^error'): #or line.startswith('(gdb)'):
break
elif line.startswith('&"break'):
pass
elif line.startswith('&"Function'):
raise GdbException(line)
elif line.startswith('&"No line'):
raise GdbException(line)
elif line.startswith('~"Make breakpoint pending on future shared'):
raise GdbException(line)
raise GdbException(
'Failed to set breakpoint.\n Output:\n {0}'.format(result)
)
def remove_all_breakpoints(self):
result = self._execute('delete')
for line in result:
if line.startswith('^done'):
return
raise GdbException(
'Failed to remove breakpoints.\n Output:\n {0}'.format(result)
)
def run_until_break(self):
result = self._execute('run', False)
for line in result:
if line.startswith('*stopped,reason="breakpoint-hit"'):
return
raise GdbException(
'Failed to run until breakpoint.\n'
)
def continue_execution_until_running(self):
result = self._execute('continue')
for line in result:
if line.startswith('*running') or line.startswith('^running'):
return
if line.startswith('*stopped,reason="breakpoint-hit"'):
continue
if line.startswith('*stopped,reason="exited-normally"'):
continue
raise GdbException(
'Failed to continue execution until running.\n'
)
def continue_execution_until_exit(self):
result = self._execute('continue', False)
for line in result:
if line.startswith('*running'):
continue
if line.startswith('*stopped,reason="breakpoint-hit"'):
continue
if (
line.startswith('*stopped,reason="exited') or
line == '*stopped\n'
):
return
raise GdbException(
'Failed to continue execution until exit.\n'
)
def continue_execution_until_error(self):
result = self._execute('continue', False)
for line in result:
if line.startswith('^error'):
return
if line.startswith('*stopped,reason="exited'):
return
if line.startswith(
'*stopped,reason="signal-received",signal-name="SIGABRT"'):
return
raise GdbException(
'Failed to continue execution until error.\n')
def continue_execution_until_break(self, ignore_count=0):
if ignore_count > 0:
result = self._execute(
'continue ' + str(ignore_count),
False
)
else:
result = self._execute('continue', False)
for line in result:
if line.startswith('*stopped,reason="breakpoint-hit"'):
return
if line.startswith('*stopped,reason="exited-normally"'):
break
raise GdbException(
'Failed to continue execution until break.\n')
def stopped_in_breakpoint(self):
while True:
line = self.get_line()
if self.verbose:
print(line)
if line.startswith('*stopped,reason="breakpoint-hit"'):
return True
return False
def quit(self):
self.proc.terminate()
# use for breakpoint, run, continue
def _execute(self, cmd, running=True):
output = []
self.proc.stdin.flush()
self.proc.stdin.write(cmd + '\n')
self.proc.stdin.flush()
sleep(1)
# look for command we just send
while True:
line = self.get_line()
if self.verbose:
print(repr(line))
if cmd not in line:
continue
else:
break
while True:
line = self.get_line()
output += [line]
if self.verbose:
print(repr(line))
if line.startswith('^done') or line.startswith('*stopped'):
break
if line.startswith('^error'):
break
if running and (line.startswith('*running') or line.startswith('^running')):
# if running and line.startswith('*running'):
break
return output