From a039fd14aa641d8427cabd9dc0caae75dd21da19 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Mon, 3 Dec 2018 13:38:45 +0300 Subject: [PATCH 01/23] add warning if waiting for START LSN during ARCHIVE backup is exceeding 30 seconds --- src/backup.c | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/backup.c b/src/backup.c index d9238929..f308aa89 100644 --- a/src/backup.c +++ b/src/backup.c @@ -1665,6 +1665,11 @@ wait_wal_lsn(XLogRecPtr lsn, bool is_start_lsn, bool wait_prev_segment) (uint32) (lsn >> 32), (uint32) lsn, wal_segment_path); } + if (!stream_wal && is_start_lsn && try_count == 30) + elog(WARNING, "By default pg_probackup assume WAL delivery method to be ARCHIVE. " + "If continius archiving is not set up, use '--stream' option to make autonomous backup. " + "Otherwise check that continius archiving works correctly."); + if (timeout > 0 && try_count > timeout) { if (file_exists) From 529173287f5937a22aaa8dc8aab628c708186242 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Mon, 3 Dec 2018 13:41:39 +0300 Subject: [PATCH 02/23] fix tabulation offset --- src/backup.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/backup.c b/src/backup.c index f308aa89..7046d1d0 100644 --- a/src/backup.c +++ b/src/backup.c @@ -1667,8 +1667,8 @@ wait_wal_lsn(XLogRecPtr lsn, bool is_start_lsn, bool wait_prev_segment) if (!stream_wal && is_start_lsn && try_count == 30) elog(WARNING, "By default pg_probackup assume WAL delivery method to be ARCHIVE. " - "If continius archiving is not set up, use '--stream' option to make autonomous backup. " - "Otherwise check that continius archiving works correctly."); + "If continius archiving is not set up, use '--stream' option to make autonomous backup. " + "Otherwise check that continius archiving works correctly."); if (timeout > 0 && try_count > timeout) { From 4ac0fab61b21d80c260078b04453f3b6a1b9e16d Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Thu, 6 Dec 2018 16:03:29 +0300 Subject: [PATCH 03/23] Show progress messages with INFO verbose level --- src/delete.c | 10 ++++++---- src/merge.c | 2 +- src/restore.c | 2 +- src/validate.c | 9 +++++---- 4 files changed, 13 insertions(+), 10 deletions(-) diff --git a/src/delete.c b/src/delete.c index e7c9ac24..8e9b965f 100644 --- a/src/delete.c +++ b/src/delete.c @@ -256,6 +256,7 @@ delete_backup_files(pgBackup *backup) char path[MAXPGPATH]; char timestamp[100]; parray *files; + size_t num_files; /* * If the backup was deleted already, there is nothing to do. @@ -286,13 +287,14 @@ delete_backup_files(pgBackup *backup) /* delete leaf node first */ parray_qsort(files, pgFileComparePathDesc); - for (i = 0; i < parray_num(files); i++) + num_files = parray_num(files); + for (i = 0; i < num_files; i++) { pgFile *file = (pgFile *) parray_get(files, i); - /* print progress */ - elog(VERBOSE, "Delete file(%zd/%lu) \"%s\"", i + 1, - (unsigned long) parray_num(files), file->path); + if (progress) + elog(INFO, "Progress: (%zd/%zd). Process file \"%s\"", + i + 1, num_files, file->path); if (remove(file->path)) { diff --git a/src/merge.c b/src/merge.c index b01a90d0..2a24b39d 100644 --- a/src/merge.c +++ b/src/merge.c @@ -385,7 +385,7 @@ merge_files(void *arg) elog(ERROR, "Interrupted during merging backups"); if (progress) - elog(LOG, "Progress: (%d/%d). Process file \"%s\"", + elog(INFO, "Progress: (%d/%d). Process file \"%s\"", i + 1, num_files, file->path); /* diff --git a/src/restore.c b/src/restore.c index 6024f8bc..8770c358 100644 --- a/src/restore.c +++ b/src/restore.c @@ -582,7 +582,7 @@ restore_files(void *arg) rel_path = GetRelativePath(file->path,from_root); if (progress) - elog(LOG, "Progress: (%d/%lu). Process file %s ", + elog(INFO, "Progress: (%d/%lu). Process file %s ", i + 1, (unsigned long) parray_num(arguments->files), rel_path); /* diff --git a/src/validate.c b/src/validate.c index 1f758d0f..edb386ef 100644 --- a/src/validate.c +++ b/src/validate.c @@ -156,9 +156,10 @@ pgBackupValidateFiles(void *arg) { int i; validate_files_arg *arguments = (validate_files_arg *)arg; + int num_files = parray_num(arguments->files); pg_crc32 crc; - for (i = 0; i < parray_num(arguments->files); i++) + for (i = 0; i < num_files; i++) { struct stat st; pgFile *file = (pgFile *) parray_get(arguments->files, i); @@ -186,9 +187,9 @@ pgBackupValidateFiles(void *arg) if (file->is_cfs) continue; - /* print progress */ - elog(VERBOSE, "Validate files: (%d/%lu) %s", - i + 1, (unsigned long) parray_num(arguments->files), file->path); + if (progress) + elog(INFO, "Progress: (%d/%d). Process file \"%s\"", + i + 1, num_files, file->path); if (stat(file->path, &st) == -1) { From e753643b8d65ad2656bca485f962de348f279433 Mon Sep 17 00:00:00 2001 From: Konstantin Knizhnik Date: Thu, 6 Dec 2018 19:00:08 +0300 Subject: [PATCH 04/23] Add error message in case of invalid compression algorithm --- src/data.c | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/data.c b/src/data.c index 19bb742b..de649103 100644 --- a/src/data.c +++ b/src/data.c @@ -98,6 +98,8 @@ do_decompress(void* dst, size_t dst_size, void const* src, size_t src_size, { case NONE_COMPRESS: case NOT_DEFINED_COMPRESS: + if (errormsg) + *errormsg = "Invalid compression algorithm"; return -1; #ifdef HAVE_LIBZ case ZLIB_COMPRESS: From 8b1b53bc3bc17d1c4894db1fa618e46e125f5980 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Thu, 6 Dec 2018 19:07:56 +0300 Subject: [PATCH 05/23] tests: test_merge_compressed_and_uncompressed_backups added --- tests/merge.py | 81 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/tests/merge.py b/tests/merge.py index 5f7ae7da..5a314dcf 100644 --- a/tests/merge.py +++ b/tests/merge.py @@ -163,6 +163,87 @@ class MergeTest(ProbackupTest, unittest.TestCase): node.cleanup() self.del_test_dir(module_name, fname) + def test_merge_compressed_and_uncompressed_backups(self): + """ + Test MERGE command with compressed and uncompressed backups + """ + fname = self.id().split(".")[3] + backup_dir = os.path.join(self.tmp_path, module_name, fname, "backup") + + # Initialize instance and backup directory + node = self.make_simple_node( + base_dir="{0}/{1}/node".format(module_name, fname), + initdb_params=["--data-checksums"], + pg_options={ + 'autovacuum': 'off' + } + ) + + self.init_pb(backup_dir) + self.add_instance(backup_dir, "node", node) + self.set_archiving(backup_dir, "node", node) + node.slow_start() + + # Fill with data + node.pgbench_init(scale=5) + + # Do compressed FULL backup + self.backup_node(backup_dir, "node", node, options=[ + '--compress-algorithm=zlib', '--stream']) + show_backup = self.show_pb(backup_dir, "node")[0] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "FULL") + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do compressed DELTA backup + self.backup_node( + backup_dir, "node", node, backup_type="delta", + options=['--compress-algorithm=zlib', '--stream']) + show_backup = self.show_pb(backup_dir, "node")[1] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "DELTA") + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do not compressed PAGE backup + self.backup_node( + backup_dir, "node", node, backup_type="page", + options=['--compress-algorithm=zlib']) + + pgdata = self.pgdata_content(node.data_dir) + + show_backup = self.show_pb(backup_dir, "node")[2] + page_id = show_backup["id"] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "PAGE") + + # Merge all backups + self.merge_backup(backup_dir, "node", page_id) + show_backups = self.show_pb(backup_dir, "node") + + self.assertEqual(len(show_backups), 1) + self.assertEqual(show_backups[0]["status"], "OK") + self.assertEqual(show_backups[0]["backup-mode"], "FULL") + + # Drop node and restore it + node.cleanup() + self.restore_node(backup_dir, 'node', node) + + pgdata_restored = self.pgdata_content(node.data_dir) + self.compare_pgdata(pgdata, pgdata_restored) + + # Clean after yourself + node.cleanup() + self.del_test_dir(module_name, fname) + # @unittest.skip("skip") def test_merge_tablespaces(self): """ From 34c05740b05dd1ee89745d9b4678e1220ed6b812 Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Thu, 6 Dec 2018 19:41:10 +0300 Subject: [PATCH 06/23] Restore to_backup sizes after copying metadata from from_backup --- src/merge.c | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/src/merge.c b/src/merge.c index 2a24b39d..6d6d978e 100644 --- a/src/merge.c +++ b/src/merge.c @@ -168,6 +168,8 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup) merge_files_arg *threads_args = NULL; int i; bool merge_isok = true; + int64 to_data_bytes, + to_wal_bytes; elog(INFO, "Merging backup %s with backup %s", from_backup_id, to_backup_id); @@ -282,26 +284,28 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup) */ to_backup->status = BACKUP_STATUS_OK; /* Compute summary of size of regular files in the backup */ - to_backup->data_bytes = 0; + to_data_bytes = 0; for (i = 0; i < parray_num(files); i++) { pgFile *file = (pgFile *) parray_get(files, i); if (S_ISDIR(file->mode)) - to_backup->data_bytes += 4096; + to_data_bytes += 4096; /* Count the amount of the data actually copied */ else if (S_ISREG(file->mode)) - to_backup->data_bytes += file->write_size; + to_data_bytes += file->write_size; } /* compute size of wal files of this backup stored in the archive */ if (!to_backup->stream) - to_backup->wal_bytes = instance_config.xlog_seg_size * + to_wal_bytes = instance_config.xlog_seg_size * (to_backup->stop_lsn / instance_config.xlog_seg_size - to_backup->start_lsn / instance_config.xlog_seg_size + 1); else - to_backup->wal_bytes = BYTES_INVALID; + to_wal_bytes = BYTES_INVALID; write_backup_filelist(to_backup, files, from_database_path); + to_backup->data_bytes = to_data_bytes; + to_backup->wal_bytes = to_wal_bytes; write_backup(to_backup); delete_source_backup: @@ -314,6 +318,7 @@ delete_source_backup: /* * Delete files which are not in from_backup file list. */ + parray_qsort(files, pgFileComparePathDesc); for (i = 0; i < parray_num(to_files); i++) { pgFile *file = (pgFile *) parray_get(to_files, i); @@ -341,6 +346,9 @@ delete_source_backup: to_backup->backup_mode = BACKUP_MODE_FULL; to_backup->status = BACKUP_STATUS_OK; to_backup->parent_backup = INVALID_BACKUP_ID; + /* Restore sizes */ + to_backup->data_bytes = to_data_bytes; + to_backup->wal_bytes = to_wal_bytes; write_backup(to_backup); /* Cleanup */ From 59ab21e5c2be7c311b8474a714ff6880c398e790 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Fri, 7 Dec 2018 09:43:38 +0300 Subject: [PATCH 07/23] tests: test_merge_compressed_backups_1 added --- tests/merge.py | 86 ++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 79 insertions(+), 7 deletions(-) diff --git a/tests/merge.py b/tests/merge.py index 5a314dcf..cbe84d5c 100644 --- a/tests/merge.py +++ b/tests/merge.py @@ -163,6 +163,83 @@ class MergeTest(ProbackupTest, unittest.TestCase): node.cleanup() self.del_test_dir(module_name, fname) + def test_merge_compressed_backups_1(self): + """ + Test MERGE command with compressed backups + """ + fname = self.id().split(".")[3] + backup_dir = os.path.join(self.tmp_path, module_name, fname, "backup") + + # Initialize instance and backup directory + node = self.make_simple_node( + base_dir="{0}/{1}/node".format(module_name, fname), + initdb_params=["--data-checksums"], + pg_options={ + 'autovacuum': 'off' + } + ) + + self.init_pb(backup_dir) + self.add_instance(backup_dir, "node", node) + self.set_archiving(backup_dir, "node", node) + node.slow_start() + + # Fill with data + node.pgbench_init(scale=5) + + # Do compressed FULL backup + self.backup_node(backup_dir, "node", node, options=[ + '--compress-algorithm=zlib', '--stream']) + show_backup = self.show_pb(backup_dir, "node")[0] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "FULL") + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do compressed DELTA backup + self.backup_node( + backup_dir, "node", node, backup_type="delta", + options=['--compress-algorithm=zlib', '--stream']) + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do compressed PAGE backup + self.backup_node( + backup_dir, "node", node, backup_type="page", + options=['--compress-algorithm=zlib']) + + pgdata = self.pgdata_content(node.data_dir) + + show_backup = self.show_pb(backup_dir, "node")[2] + page_id = show_backup["id"] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "PAGE") + + # Merge all backups + self.merge_backup(backup_dir, "node", page_id) + show_backups = self.show_pb(backup_dir, "node") + + self.assertEqual(len(show_backups), 1) + self.assertEqual(show_backups[0]["status"], "OK") + self.assertEqual(show_backups[0]["backup-mode"], "FULL") + + # Drop node and restore it + node.cleanup() + self.restore_node(backup_dir, 'node', node) + + pgdata_restored = self.pgdata_content(node.data_dir) + self.compare_pgdata(pgdata, pgdata_restored) + + # Clean after yourself + node.cleanup() + self.del_test_dir(module_name, fname) + def test_merge_compressed_and_uncompressed_backups(self): """ Test MERGE command with compressed and uncompressed backups @@ -203,19 +280,14 @@ class MergeTest(ProbackupTest, unittest.TestCase): self.backup_node( backup_dir, "node", node, backup_type="delta", options=['--compress-algorithm=zlib', '--stream']) - show_backup = self.show_pb(backup_dir, "node")[1] - - self.assertEqual(show_backup["status"], "OK") - self.assertEqual(show_backup["backup-mode"], "DELTA") # Change data pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) pgbench.wait() - # Do not compressed PAGE backup + # Do uncompressed PAGE backup self.backup_node( - backup_dir, "node", node, backup_type="page", - options=['--compress-algorithm=zlib']) + backup_dir, "node", node, backup_type="page") pgdata = self.pgdata_content(node.data_dir) From 2837399e6bdb9ccde4a28e68aa3e9757af4dabb5 Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Fri, 7 Dec 2018 17:24:21 +0300 Subject: [PATCH 08/23] Fix MERGE of compressed and uncompressed backups --- src/merge.c | 116 ++++++++++++++++++++++--------------------------- tests/merge.py | 2 - 2 files changed, 51 insertions(+), 67 deletions(-) diff --git a/src/merge.c b/src/merge.c index 6d6d978e..e0669586 100644 --- a/src/merge.c +++ b/src/merge.c @@ -131,7 +131,6 @@ do_merge(time_t backup_id) { pgBackup *from_backup = (pgBackup *) parray_get(backups, i - 1); - full_backup = (pgBackup *) parray_get(backups, i); merge_backups(full_backup, from_backup); } @@ -168,8 +167,6 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup) merge_files_arg *threads_args = NULL; int i; bool merge_isok = true; - int64 to_data_bytes, - to_wal_bytes; elog(INFO, "Merging backup %s with backup %s", from_backup_id, to_backup_id); @@ -283,29 +280,37 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup) * Update to_backup metadata. */ to_backup->status = BACKUP_STATUS_OK; + to_backup->parent_backup = INVALID_BACKUP_ID; + to_backup->start_lsn = from_backup->start_lsn; + to_backup->stop_lsn = from_backup->stop_lsn; + to_backup->recovery_time = from_backup->recovery_time; + to_backup->recovery_xid = from_backup->recovery_xid; + /* + * If one of the backups isn't "stream" backup then the target backup become + * non-stream backup too. + */ + to_backup->stream = to_backup->stream && from_backup->stream; /* Compute summary of size of regular files in the backup */ - to_data_bytes = 0; + to_backup->data_bytes = 0; for (i = 0; i < parray_num(files); i++) { pgFile *file = (pgFile *) parray_get(files, i); if (S_ISDIR(file->mode)) - to_data_bytes += 4096; + to_backup->data_bytes += 4096; /* Count the amount of the data actually copied */ else if (S_ISREG(file->mode)) - to_data_bytes += file->write_size; + to_backup->data_bytes += file->write_size; } /* compute size of wal files of this backup stored in the archive */ if (!to_backup->stream) - to_wal_bytes = instance_config.xlog_seg_size * + to_backup->wal_bytes = instance_config.xlog_seg_size * (to_backup->stop_lsn / instance_config.xlog_seg_size - to_backup->start_lsn / instance_config.xlog_seg_size + 1); else - to_wal_bytes = BYTES_INVALID; + to_backup->wal_bytes = BYTES_INVALID; write_backup_filelist(to_backup, files, from_database_path); - to_backup->data_bytes = to_data_bytes; - to_backup->wal_bytes = to_wal_bytes; write_backup(to_backup); delete_source_backup: @@ -330,27 +335,6 @@ delete_source_backup: } } - /* - * Rename FULL backup directory. - */ - elog(INFO, "Rename %s to %s", to_backup_id, from_backup_id); - if (rename(to_backup_path, from_backup_path) == -1) - elog(ERROR, "Could not rename directory \"%s\" to \"%s\": %s", - to_backup_path, from_backup_path, strerror(errno)); - - /* - * Merging finished, now we can safely update ID of the destination backup. - */ - pgBackupCopy(to_backup, from_backup); - /* Correct metadata */ - to_backup->backup_mode = BACKUP_MODE_FULL; - to_backup->status = BACKUP_STATUS_OK; - to_backup->parent_backup = INVALID_BACKUP_ID; - /* Restore sizes */ - to_backup->data_bytes = to_data_bytes; - to_backup->wal_bytes = to_wal_bytes; - write_backup(to_backup); - /* Cleanup */ if (threads) { @@ -384,6 +368,8 @@ merge_files(void *arg) for (i = 0; i < num_files; i++) { pgFile *file = (pgFile *) parray_get(argument->files, i); + pgFile *to_file; + pgFile **res_file; if (!pg_atomic_test_set_flag(&file->lock)) continue; @@ -392,17 +378,24 @@ merge_files(void *arg) if (interrupted) elog(ERROR, "Interrupted during merging backups"); + /* Directories were created before */ + if (S_ISDIR(file->mode)) + continue; + if (progress) elog(INFO, "Progress: (%d/%d). Process file \"%s\"", i + 1, num_files, file->path); + res_file = parray_bsearch(argument->to_files, file, + pgFileComparePathDesc); + to_file = (res_file) ? *res_file : NULL; + /* * Skip files which haven't changed since previous backup. But in case * of DELTA backup we should consider n_blocks to truncate the target * backup. */ - if (file->write_size == BYTES_INVALID && - file->n_blocks == -1) + if (file->write_size == BYTES_INVALID && file->n_blocks == -1) { elog(VERBOSE, "Skip merging file \"%s\", the file didn't change", file->path); @@ -411,27 +404,16 @@ merge_files(void *arg) * If the file wasn't changed in PAGE backup, retreive its * write_size from previous FULL backup. */ - if (S_ISREG(file->mode)) + if (to_file) { - pgFile **res_file; - - res_file = parray_bsearch(argument->to_files, file, - pgFileComparePathDesc); - if (res_file && *res_file) - { - file->compress_alg = (*res_file)->compress_alg; - file->write_size = (*res_file)->write_size; - file->crc = (*res_file)->crc; - } + file->compress_alg = to_file->compress_alg; + file->write_size = to_file->write_size; + file->crc = to_file->crc; } continue; } - /* Directories were created before */ - if (S_ISDIR(file->mode)) - continue; - /* * Move the file. We need to decompress it and compress again if * necessary. @@ -447,7 +429,7 @@ merge_files(void *arg) file->path + to_root_len + 1); /* - * We need more complicate algorithm if target file exists and it is + * We need more complicate algorithm if target file should be * compressed. */ if (to_backup->compress_alg == PGLZ_COMPRESS || @@ -462,33 +444,36 @@ merge_files(void *arg) /* * Merge files: - * - decompress first file - * - decompress second file and merge with first decompressed file + * - if target file exists restore and decompress it to the temp + * path + * - decompress source file if necessary and merge it with the + * target decompressed file * - compress result file */ - elog(VERBOSE, "File is compressed, decompress to the temporary file \"%s\"", - tmp_file_path); - - prev_path = file->path; /* - * We need to decompress target file only if it exists. + * We need to decompress target file if it exists. */ - if (fileExists(to_path_tmp)) + if (to_file) { + elog(VERBOSE, "Merge target and source files into the temporary path \"%s\"", + tmp_file_path); + /* * file->path points to the file in from_root directory. But we * need the file in directory to_root. */ - file->path = to_path_tmp; - - /* Decompress first/target file */ - restore_data_file(tmp_file_path, file, false, false, + prev_path = to_file->path; + to_file->path = to_path_tmp; + /* Decompress target file into temporary one */ + restore_data_file(tmp_file_path, to_file, false, false, parse_program_version(to_backup->program_version)); - - file->path = prev_path; + to_file->path = prev_path; } - /* Merge second/source file with first/target file */ + else + elog(VERBOSE, "Restore source file into the temporary path \"%s\"", + tmp_file_path); + /* Merge source file with target file */ restore_data_file(tmp_file_path, file, from_backup->backup_mode == BACKUP_MODE_DIFF_DELTA, false, @@ -497,7 +482,8 @@ merge_files(void *arg) elog(VERBOSE, "Compress file and save it to the directory \"%s\"", argument->to_root); - /* Again we need change path */ + /* Again we need to change path */ + prev_path = file->path; file->path = tmp_file_path; /* backup_data_file() requires file size to calculate nblocks */ file->size = pgFileSize(file->path); diff --git a/tests/merge.py b/tests/merge.py index cbe84d5c..c3b3169a 100644 --- a/tests/merge.py +++ b/tests/merge.py @@ -875,8 +875,6 @@ class MergeTest(ProbackupTest, unittest.TestCase): repr(self.output), self.cmd)) except ProbackupException as e: self.assertTrue( - "WARNING: Backup {0} data files are corrupted".format( - backup_id) in e.message and "ERROR: Merging of backup {0} failed".format( backup_id) in e.message, '\n Unexpected Error Message: {0}\n CMD: {1}'.format( From b455420a18b1470726c3b44714cbc74800d40df2 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Fri, 7 Dec 2018 23:39:39 +0300 Subject: [PATCH 09/23] tests: more merge tests added --- tests/merge.py | 153 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) diff --git a/tests/merge.py b/tests/merge.py index c3b3169a..e4209441 100644 --- a/tests/merge.py +++ b/tests/merge.py @@ -316,6 +316,159 @@ class MergeTest(ProbackupTest, unittest.TestCase): node.cleanup() self.del_test_dir(module_name, fname) + def test_merge_compressed_and_uncompressed_backups_1(self): + """ + Test MERGE command with compressed and uncompressed backups + """ + fname = self.id().split(".")[3] + backup_dir = os.path.join(self.tmp_path, module_name, fname, "backup") + + # Initialize instance and backup directory + node = self.make_simple_node( + base_dir="{0}/{1}/node".format(module_name, fname), + initdb_params=["--data-checksums"], + pg_options={ + 'autovacuum': 'off' + } + ) + + self.init_pb(backup_dir) + self.add_instance(backup_dir, "node", node) + self.set_archiving(backup_dir, "node", node) + node.slow_start() + + # Fill with data + node.pgbench_init(scale=5) + + # Do compressed FULL backup + self.backup_node(backup_dir, "node", node, options=[ + '--compress-algorithm=zlib', '--stream']) + show_backup = self.show_pb(backup_dir, "node")[0] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "FULL") + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do uncompressed DELTA backup + self.backup_node( + backup_dir, "node", node, backup_type="delta", + options=['--stream']) + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do compressed PAGE backup + self.backup_node( + backup_dir, "node", node, backup_type="page", + options=['--compress-algorithm=zlib']) + + pgdata = self.pgdata_content(node.data_dir) + + show_backup = self.show_pb(backup_dir, "node")[2] + page_id = show_backup["id"] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "PAGE") + + # Merge all backups + self.merge_backup(backup_dir, "node", page_id) + show_backups = self.show_pb(backup_dir, "node") + + self.assertEqual(len(show_backups), 1) + self.assertEqual(show_backups[0]["status"], "OK") + self.assertEqual(show_backups[0]["backup-mode"], "FULL") + + # Drop node and restore it + node.cleanup() + self.restore_node(backup_dir, 'node', node) + + pgdata_restored = self.pgdata_content(node.data_dir) + self.compare_pgdata(pgdata, pgdata_restored) + + # Clean after yourself + node.cleanup() + self.del_test_dir(module_name, fname) + + def test_merge_compressed_and_uncompressed_backups_2(self): + """ + Test MERGE command with compressed and uncompressed backups + """ + fname = self.id().split(".")[3] + backup_dir = os.path.join(self.tmp_path, module_name, fname, "backup") + + # Initialize instance and backup directory + node = self.make_simple_node( + base_dir="{0}/{1}/node".format(module_name, fname), + initdb_params=["--data-checksums"], + pg_options={ + 'autovacuum': 'off' + } + ) + + self.init_pb(backup_dir) + self.add_instance(backup_dir, "node", node) + self.set_archiving(backup_dir, "node", node) + node.slow_start() + + # Fill with data + node.pgbench_init(scale=5) + + # Do uncompressed FULL backup + self.backup_node(backup_dir, "node", node) + show_backup = self.show_pb(backup_dir, "node")[0] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "FULL") + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do compressed DELTA backup + self.backup_node( + backup_dir, "node", node, backup_type="delta", + options=['--compress-algorithm=zlib', '--stream']) + + # Change data + pgbench = node.pgbench(options=['-T', '20', '-c', '2', '--no-vacuum']) + pgbench.wait() + + # Do uncompressed PAGE backup + self.backup_node( + backup_dir, "node", node, backup_type="page") + + pgdata = self.pgdata_content(node.data_dir) + + show_backup = self.show_pb(backup_dir, "node")[2] + page_id = show_backup["id"] + + self.assertEqual(show_backup["status"], "OK") + self.assertEqual(show_backup["backup-mode"], "PAGE") + + # Merge all backups + self.merge_backup(backup_dir, "node", page_id) + show_backups = self.show_pb(backup_dir, "node") + + self.assertEqual(len(show_backups), 1) + self.assertEqual(show_backups[0]["status"], "OK") + self.assertEqual(show_backups[0]["backup-mode"], "FULL") + + # Drop node and restore it + node.cleanup() + self.restore_node(backup_dir, 'node', node) + + pgdata_restored = self.pgdata_content(node.data_dir) + self.compare_pgdata(pgdata, pgdata_restored) + + # Clean after yourself + node.cleanup() + self.del_test_dir(module_name, fname) + + # @unittest.skip("skip") def test_merge_tablespaces(self): """ From c1903db22b0812f3b327a8e3a0cba3299413d7d2 Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Mon, 10 Dec 2018 16:19:51 +0300 Subject: [PATCH 10/23] Fix format warnings --- src/archive.c | 4 ++-- src/data.c | 12 ++++++------ src/restore.c | 2 +- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/archive.c b/src/archive.c index 3333f096..774d12c4 100644 --- a/src/archive.c +++ b/src/archive.c @@ -27,7 +27,7 @@ do_archive_push(char *wal_file_path, char *wal_file_name, bool overwrite) char backup_wal_file_path[MAXPGPATH]; char absolute_wal_file_path[MAXPGPATH]; char current_dir[MAXPGPATH]; - int64 system_id; + uint64 system_id; bool is_compress = false; if (wal_file_name == NULL && wal_file_path == NULL) @@ -50,7 +50,7 @@ do_archive_push(char *wal_file_path, char *wal_file_name, bool overwrite) if(system_id != instance_config.system_identifier) elog(ERROR, "Refuse to push WAL segment %s into archive. Instance parameters mismatch." - "Instance '%s' should have SYSTEM_ID = %ld instead of %ld", + "Instance '%s' should have SYSTEM_ID = " UINT64_FORMAT " instead of " UINT64_FORMAT, wal_file_name, instance_name, instance_config.system_identifier, system_id); diff --git a/src/data.c b/src/data.c index de649103..0d0c5e5c 100644 --- a/src/data.c +++ b/src/data.c @@ -232,8 +232,8 @@ read_page_from_file(pgFile *file, BlockNumber blknum, return 0; } else - elog(WARNING, "File: %s, block %u, expected block size %d," - "but read %lu, try again", + elog(WARNING, "File: %s, block %u, expected block size %u," + "but read %zu, try again", file->path, blknum, BLCKSZ, read_len); } @@ -382,7 +382,7 @@ prepare_page(backup_files_arg *arguments, else if (page_size != BLCKSZ) { free(ptrack_page); - elog(ERROR, "File: %s, block %u, expected block size %d, but read %lu", + elog(ERROR, "File: %s, block %u, expected block size %d, but read %zu", file->path, absolute_blknum, BLCKSZ, page_size); } else @@ -574,7 +574,7 @@ backup_data_file(backup_files_arg* arguments, if (file->size % BLCKSZ != 0) { fclose(in); - elog(ERROR, "File: %s, invalid file size %lu", file->path, file->size); + elog(ERROR, "File: %s, invalid file size %zu", file->path, file->size); } /* @@ -789,7 +789,7 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate, read_len = fread(compressed_page.data, 1, MAXALIGN(header.compressed_size), in); if (read_len != MAXALIGN(header.compressed_size)) - elog(ERROR, "cannot read block %u of \"%s\" read %lu of %d", + elog(ERROR, "cannot read block %u of \"%s\" read %zu of %d", blknum, file->path, read_len, header.compressed_size); /* @@ -1620,7 +1620,7 @@ check_file_pages(pgFile *file, XLogRecPtr stop_lsn, uint32 checksum_version, read_len = fread(compressed_page.data, 1, MAXALIGN(header.compressed_size), in); if (read_len != MAXALIGN(header.compressed_size)) - elog(ERROR, "cannot read block %u of \"%s\" read %lu of %d", + elog(ERROR, "cannot read block %u of \"%s\" read %zu of %d", blknum, file->path, read_len, header.compressed_size); COMP_FILE_CRC32(use_crc32c, crc, compressed_page.data, read_len); diff --git a/src/restore.c b/src/restore.c index 8770c358..e3591e06 100644 --- a/src/restore.c +++ b/src/restore.c @@ -480,7 +480,7 @@ restore_backup(pgBackup *backup) /* By default there are some error */ threads_args[i].ret = 1; - elog(LOG, "Start thread for num:%li", parray_num(files)); + elog(LOG, "Start thread for num:%zu", parray_num(files)); pthread_create(&threads[i], NULL, restore_files, arg); } From dd18928e1984927edafe175a2a1bb30f9f64d66e Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Mon, 10 Dec 2018 18:02:22 +0300 Subject: [PATCH 11/23] Make compiler happy --- src/utils/logger.c | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/src/utils/logger.c b/src/utils/logger.c index fa1c0039..a4f2d692 100644 --- a/src/utils/logger.c +++ b/src/utils/logger.c @@ -232,18 +232,23 @@ elog_internal(int elevel, bool file_only, const char *fmt, va_list args) * Write to stderr if the message was not written to log file. * Write to stderr if the message level is greater than WARNING anyway. */ - if (write_to_stderr) + if (write_to_stderr && write_to_file) { write_elevel(stderr, elevel); - if (write_to_file) - vfprintf(stderr, fmt, std_args); - else - vfprintf(stderr, fmt, args); + + vfprintf(stderr, fmt, std_args); fputc('\n', stderr); fflush(stderr); - if (write_to_file) - va_end(std_args); + va_end(std_args); + } + else if (write_to_stderr) + { + write_elevel(stderr, elevel); + + vfprintf(stderr, fmt, args); + fputc('\n', stderr); + fflush(stderr); } exit_if_necessary(elevel); From 9ac30ada5443e97de9152de7bab5c3b9752b2abb Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Mon, 10 Dec 2018 18:49:39 +0300 Subject: [PATCH 12/23] bugfix: correctly log cmdline to logfile --- src/pg_probackup.c | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/src/pg_probackup.c b/src/pg_probackup.c index 17676876..04b5d486 100644 --- a/src/pg_probackup.c +++ b/src/pg_probackup.c @@ -331,15 +331,6 @@ main(int argc, char *argv[]) if (rc != -1 && !S_ISDIR(stat_buf.st_mode)) elog(ERROR, "-B, --backup-path must be a path to directory"); - /* command was initialized for a few commands */ - if (command) - { - elog_file(INFO, "command: %s", command); - - pfree(command); - command = NULL; - } - /* Option --instance is required for all commands except init and show */ if (backup_subcmd != INIT_CMD && backup_subcmd != SHOW_CMD && backup_subcmd != VALIDATE_CMD) @@ -390,6 +381,15 @@ main(int argc, char *argv[]) /* Initialize logger */ init_logger(backup_path, &instance_config.logger); + /* command was initialized for a few commands */ + if (command) + { + elog_file(INFO, "command: %s", command); + + pfree(command); + command = NULL; + } + /* * We have read pgdata path from command line or from configuration file. * Ensure that pgdata is an absolute path. From eead6b6307b464b67e567516609baac7fce7560c Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Mon, 10 Dec 2018 19:00:37 +0300 Subject: [PATCH 13/23] Free unnecessary file object --- src/dir.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/dir.c b/src/dir.c index 710598b7..ae8cb20b 100644 --- a/src/dir.c +++ b/src/dir.c @@ -443,6 +443,9 @@ dir_list_file(parray *files, const char *root, bool exclude, bool omit_symlink, parray_append(files, file); dir_list_file_internal(files, root, file, exclude, omit_symlink, black_list); + + if (!add_root) + pgFileFree(file); } /* From 385f979fe7daf91001eea4f4c123807bb6f576ee Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Tue, 11 Dec 2018 12:27:36 +0300 Subject: [PATCH 14/23] Make compiler happy again --- src/utils/logger.c | 98 ++++++++++++++++++++++++++-------------------- 1 file changed, 55 insertions(+), 43 deletions(-) diff --git a/src/utils/logger.c b/src/utils/logger.c index a4f2d692..ba054a62 100644 --- a/src/utils/logger.c +++ b/src/utils/logger.c @@ -40,10 +40,10 @@ typedef enum void pg_log(eLogType type, const char *fmt,...) pg_attribute_printf(2, 3); -static void elog_internal(int elevel, bool file_only, const char *fmt, va_list args) - pg_attribute_printf(3, 0); +static void elog_internal(int elevel, bool file_only, const char *message); static void elog_stderr(int elevel, const char *fmt, ...) pg_attribute_printf(2, 3); +static char *get_log_message(const char *fmt, va_list args) pg_attribute_printf(1, 0); /* Functions to work with log files */ static void open_logfile(FILE **file, const char *filename_format); @@ -74,7 +74,7 @@ init_logger(const char *root_path, LoggerConfig *config) /* Set log path */ if (config->log_directory == NULL) { - config->log_directory = palloc(MAXPGPATH); + config->log_directory = pgut_malloc(MAXPGPATH); join_path_components(config->log_directory, root_path, LOG_DIRECTORY_DEFAULT); } @@ -148,13 +148,11 @@ exit_if_necessary(int elevel) * Actual implementation for elog() and pg_log(). */ static void -elog_internal(int elevel, bool file_only, const char *fmt, va_list args) +elog_internal(int elevel, bool file_only, const char *message) { bool write_to_file, write_to_error_log, write_to_stderr; - va_list error_args, - std_args; time_t log_time = (time_t) time(NULL); char strfbuf[128]; @@ -165,22 +163,8 @@ elog_internal(int elevel, bool file_only, const char *fmt, va_list args) write_to_stderr = elevel >= logger_config.log_level_console && !file_only; pthread_lock(&log_file_mutex); -#ifdef WIN32 - std_args = NULL; - error_args = NULL; -#endif loggin_in_progress = true; - /* We need copy args only if we need write to error log file */ - if (write_to_error_log) - va_copy(error_args, args); - /* - * We need copy args only if we need write to stderr. But do not copy args - * if we need to log only to stderr. - */ - if (write_to_stderr && write_to_file) - va_copy(std_args, args); - if (write_to_file || write_to_error_log) strftime(strfbuf, sizeof(strfbuf), "%Y-%m-%d %H:%M:%S %Z", localtime(&log_time)); @@ -203,8 +187,7 @@ elog_internal(int elevel, bool file_only, const char *fmt, va_list args) fprintf(log_file, "%s: ", strfbuf); write_elevel(log_file, elevel); - vfprintf(log_file, fmt, args); - fputc('\n', log_file); + fprintf(log_file, "%s\n", message); fflush(log_file); } @@ -221,33 +204,19 @@ elog_internal(int elevel, bool file_only, const char *fmt, va_list args) fprintf(error_log_file, "%s: ", strfbuf); write_elevel(error_log_file, elevel); - vfprintf(error_log_file, fmt, error_args); - fputc('\n', error_log_file); + fprintf(error_log_file, "%s\n", message); fflush(error_log_file); - - va_end(error_args); } /* * Write to stderr if the message was not written to log file. * Write to stderr if the message level is greater than WARNING anyway. */ - if (write_to_stderr && write_to_file) + if (write_to_stderr) { write_elevel(stderr, elevel); - vfprintf(stderr, fmt, std_args); - fputc('\n', stderr); - fflush(stderr); - - va_end(std_args); - } - else if (write_to_stderr) - { - write_elevel(stderr, elevel); - - vfprintf(stderr, fmt, args); - fputc('\n', stderr); + fprintf(stderr, "%s\n", message); fflush(stderr); } @@ -285,12 +254,44 @@ elog_stderr(int elevel, const char *fmt, ...) exit_if_necessary(elevel); } +/* + * Formats text data under the control of fmt and returns it in an allocated + * buffer. + */ +static char * +get_log_message(const char *fmt, va_list args) +{ + size_t len = 256; /* initial assumption about buffer size */ + + for (;;) + { + char *result; + size_t newlen; + va_list copy_args; + + result = (char *) pgut_malloc(len); + + /* Try to format the data */ + va_copy(copy_args, args); + newlen = pvsnprintf(result, len, fmt, copy_args); + va_end(copy_args); + + if (newlen < len) + return result; /* success */ + + /* Release buffer and loop around to try again with larger len. */ + pfree(result); + len = newlen; + } +} + /* * Logs to stderr or to log file and exit if ERROR. */ void elog(int elevel, const char *fmt, ...) { + char *message; va_list args; /* @@ -302,8 +303,11 @@ elog(int elevel, const char *fmt, ...) return; va_start(args, fmt); - elog_internal(elevel, false, fmt, args); + message = get_log_message(fmt, args); va_end(args); + + elog_internal(elevel, false, message); + pfree(message); } /* @@ -312,6 +316,7 @@ elog(int elevel, const char *fmt, ...) void elog_file(int elevel, const char *fmt, ...) { + char *message; va_list args; /* @@ -322,8 +327,11 @@ elog_file(int elevel, const char *fmt, ...) return; va_start(args, fmt); - elog_internal(elevel, true, fmt, args); + message = get_log_message(fmt, args); va_end(args); + + elog_internal(elevel, true, message); + pfree(message); } /* @@ -332,6 +340,7 @@ elog_file(int elevel, const char *fmt, ...) void pg_log(eLogType type, const char *fmt, ...) { + char *message; va_list args; int elevel = INFO; @@ -364,8 +373,11 @@ pg_log(eLogType type, const char *fmt, ...) return; va_start(args, fmt); - elog_internal(elevel, false, fmt, args); + message = get_log_message(fmt, args); va_end(args); + + elog_internal(elevel, false, message); + pfree(message); } /* @@ -450,7 +462,7 @@ logfile_getname(const char *format, time_t timestamp) logger_config.log_directory[0] == '\0') elog_stderr(ERROR, "logging path is not set"); - filename = (char *) palloc(MAXPGPATH); + filename = (char *) pgut_malloc(MAXPGPATH); snprintf(filename, MAXPGPATH, "%s/", logger_config.log_directory); From eed59874069c05026ae76c04355ead8f9795d9d9 Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Tue, 11 Dec 2018 12:47:21 +0300 Subject: [PATCH 15/23] Remove BOM marker from configuration.c --- src/utils/configuration.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/utils/configuration.c b/src/utils/configuration.c index 9b88d159..fe50c494 100644 --- a/src/utils/configuration.c +++ b/src/utils/configuration.c @@ -1,4 +1,4 @@ -/*------------------------------------------------------------------------- +/*------------------------------------------------------------------------- * * configuration.c: - function implementations to work with pg_probackup * configurations. From 30cf45816d27b2bb877bc54689f71f09483dc004 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Wed, 12 Dec 2018 00:29:20 +0300 Subject: [PATCH 16/23] check backup version before backup validation, throw an error if backup version > binary version --- src/validate.c | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/validate.c b/src/validate.c index edb386ef..14d347ca 100644 --- a/src/validate.c +++ b/src/validate.c @@ -52,6 +52,14 @@ pgBackupValidate(pgBackup *backup) validate_files_arg *threads_args; int i; + /* Check backup version */ + if (backup->program_version && + parse_program_version(backup->program_version) > parse_program_version(PROGRAM_VERSION)) + elog(ERROR, "pg_probackup binary version is %s, but backup %s version is %s. " + "pg_probackup do not guarantee to be forward compatible. " + "Please upgrade pg_probackup binary.", + PROGRAM_VERSION, base36enc(backup->start_time), backup->program_version); + /* Revalidation is attempted for DONE, ORPHAN and CORRUPT backups */ if (backup->status != BACKUP_STATUS_OK && backup->status != BACKUP_STATUS_DONE && From 23d1a26e024c3fb66f4d5c514abd80ed53558cae Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Wed, 12 Dec 2018 15:10:25 +0300 Subject: [PATCH 17/23] Fix MERGE of uncompressed with compressed backups --- src/catalog.c | 14 -------------- src/data.c | 23 +++++++++++++++++------ src/dir.c | 2 +- src/merge.c | 23 ++++++++++++++++++++++- src/pg_probackup.h | 1 - 5 files changed, 40 insertions(+), 23 deletions(-) diff --git a/src/catalog.c b/src/catalog.c index 77ac0f34..b30fcc66 100644 --- a/src/catalog.c +++ b/src/catalog.c @@ -820,20 +820,6 @@ pgBackupInit(pgBackup *backup) backup->server_version[0] = '\0'; } -/* - * Copy backup metadata from **src** into **dst**. - */ -void -pgBackupCopy(pgBackup *dst, pgBackup *src) -{ - pfree(dst->primary_conninfo); - - memcpy(dst, src, sizeof(pgBackup)); - - if (src->primary_conninfo) - dst->primary_conninfo = pstrdup(src->primary_conninfo); -} - /* free pgBackup object */ void pgBackupFree(void *backup) diff --git a/src/data.c b/src/data.c index 0d0c5e5c..ccbe9ca8 100644 --- a/src/data.c +++ b/src/data.c @@ -829,6 +829,8 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate, if (write_header) { + /* We uncompressed the page, so its size is BLCKSZ */ + header.compressed_size = BLCKSZ; if (fwrite(&header, 1, sizeof(header), out) != sizeof(header)) elog(ERROR, "cannot write header of block %u of \"%s\": %s", blknum, file->path, strerror(errno)); @@ -1592,19 +1594,23 @@ check_file_pages(pgFile *file, XLogRecPtr stop_lsn, uint32 checksum_version, if (read_len == 0 && feof(in)) break; /* EOF found */ else if (read_len != 0 && feof(in)) - elog(ERROR, + elog(WARNING, "odd size page found at block %u of \"%s\"", blknum, file->path); else - elog(ERROR, "cannot read header of block %u of \"%s\": %s", + elog(WARNING, "cannot read header of block %u of \"%s\": %s", blknum, file->path, strerror(errno_tmp)); + return false; } COMP_FILE_CRC32(use_crc32c, crc, &header, read_len); if (header.block < blknum) - elog(ERROR, "backup is broken at file->path %s block %u", + { + elog(WARNING, "backup is broken at file->path %s block %u", file->path, blknum); + return false; + } blknum = header.block; @@ -1620,8 +1626,11 @@ check_file_pages(pgFile *file, XLogRecPtr stop_lsn, uint32 checksum_version, read_len = fread(compressed_page.data, 1, MAXALIGN(header.compressed_size), in); if (read_len != MAXALIGN(header.compressed_size)) - elog(ERROR, "cannot read block %u of \"%s\" read %zu of %d", + { + elog(WARNING, "cannot read block %u of \"%s\" read %zu of %d", blknum, file->path, read_len, header.compressed_size); + return false; + } COMP_FILE_CRC32(use_crc32c, crc, compressed_page.data, read_len); @@ -1648,11 +1657,13 @@ check_file_pages(pgFile *file, XLogRecPtr stop_lsn, uint32 checksum_version, is_valid = false; continue; } - elog(ERROR, "page of file \"%s\" uncompressed to %d bytes. != BLCKSZ", + elog(WARNING, "page of file \"%s\" uncompressed to %d bytes. != BLCKSZ", file->path, uncompressed_size); + return false; } + if (validate_one_page(page.data, file, blknum, - stop_lsn, checksum_version) == PAGE_IS_FOUND_AND_NOT_VALID) + stop_lsn, checksum_version) == PAGE_IS_FOUND_AND_NOT_VALID) is_valid = false; } else diff --git a/src/dir.c b/src/dir.c index ae8cb20b..386f49bc 100644 --- a/src/dir.c +++ b/src/dir.c @@ -211,7 +211,7 @@ pgFileInit(const char *path) strcpy(file->path, path); /* enough buffer size guaranteed */ /* Get file name from the path */ - file_name = strrchr(file->path, '/'); + file_name = last_dir_separator(file->path); if (file_name == NULL) file->name = file->path; else diff --git a/src/merge.c b/src/merge.c index e0669586..b936469b 100644 --- a/src/merge.c +++ b/src/merge.c @@ -91,7 +91,8 @@ do_merge(time_t backup_id) } else { - Assert(dest_backup); + if (dest_backup == NULL) + elog(ERROR, "Target backup %s was not found", base36enc(backup_id)); if (backup->start_time != prev_parent) continue; @@ -335,6 +336,20 @@ delete_source_backup: } } + /* + * Rename FULL backup directory. + */ + elog(INFO, "Rename %s to %s", to_backup_id, from_backup_id); + if (rename(to_backup_path, from_backup_path) == -1) + elog(ERROR, "Could not rename directory \"%s\" to \"%s\": %s", + to_backup_path, from_backup_path, strerror(errno)); + + /* + * Merging finished, now we can safely update ID of the destination backup. + */ + to_backup->start_time = from_backup->start_time; + write_backup(to_backup); + /* Cleanup */ if (threads) { @@ -526,6 +541,12 @@ merge_files(void *arg) else copy_file(argument->from_root, argument->to_root, file); + /* + * We need to save compression algorithm type of the target backup to be + * able to restore in the future. + */ + file->compress_alg = to_backup->compress_alg; + if (file->write_size != BYTES_INVALID) elog(LOG, "Moved file \"%s\": " INT64_FORMAT " bytes", file->path, file->write_size); diff --git a/src/pg_probackup.h b/src/pg_probackup.h index 0dc649a2..546c1016 100644 --- a/src/pg_probackup.h +++ b/src/pg_probackup.h @@ -476,7 +476,6 @@ extern void pgBackupGetPath2(const pgBackup *backup, char *path, size_t len, const char *subdir1, const char *subdir2); extern int pgBackupCreateDir(pgBackup *backup); extern void pgBackupInit(pgBackup *backup); -extern void pgBackupCopy(pgBackup *dst, pgBackup *src); extern void pgBackupFree(void *backup); extern int pgBackupCompareId(const void *f1, const void *f2); extern int pgBackupCompareIdDesc(const void *f1, const void *f2); From 0b05cf0343a688178fafa33ad916c81875287234 Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Wed, 12 Dec 2018 17:27:31 +0300 Subject: [PATCH 18/23] Improve comment within on_interrupt() --- src/utils/pgut.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/utils/pgut.c b/src/utils/pgut.c index ccf832b9..e08b53ea 100644 --- a/src/utils/pgut.c +++ b/src/utils/pgut.c @@ -703,7 +703,10 @@ on_interrupt(void) /* Set interruped flag */ interrupted = true; - /* User promts password, call on_cleanup() byhand */ + /* + * User promts password, call on_cleanup() byhand. If we don't do that we + * will stuck forever until an user enters a password. + */ if (in_password) { on_cleanup(); From 5ad509a40efec5af6ce22c5cd8c08b2dee94147b Mon Sep 17 00:00:00 2001 From: Arthur Zakirov Date: Wed, 12 Dec 2018 17:52:01 +0300 Subject: [PATCH 19/23] Fix comment introduced in 0b05cf0343 --- src/utils/pgut.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/utils/pgut.c b/src/utils/pgut.c index e08b53ea..a7e12e91 100644 --- a/src/utils/pgut.c +++ b/src/utils/pgut.c @@ -704,8 +704,8 @@ on_interrupt(void) interrupted = true; /* - * User promts password, call on_cleanup() byhand. If we don't do that we - * will stuck forever until an user enters a password. + * User promts password, call on_cleanup() byhand. Unless we do that we will + * get stuck forever until a user enters a password. */ if (in_password) { From 9484b6de27c3768e7105a87f5c94e989611767fc Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Wed, 12 Dec 2018 20:32:42 +0300 Subject: [PATCH 20/23] tests: test_page_create_db added --- tests/page.py | 118 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 118 insertions(+) diff --git a/tests/page.py b/tests/page.py index f0647082..d0ebbb9b 100644 --- a/tests/page.py +++ b/tests/page.py @@ -1,6 +1,7 @@ import os import unittest from .helpers.ptrack_helpers import ProbackupTest, ProbackupException +from testgres import QueryException from datetime import datetime, timedelta import subprocess import gzip @@ -1030,3 +1031,120 @@ class PageBackupTest(ProbackupTest, unittest.TestCase): # Clean after yourself self.del_test_dir(module_name, fname) + + # @unittest.skip("skip") + def test_page_create_db(self): + """ + Make node, take full backup, create database db1, take page backup, + restore database and check it presense + """ + self.maxDiff = None + fname = self.id().split('.')[3] + backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup') + node = self.make_simple_node( + base_dir="{0}/{1}/node".format(module_name, fname), + set_replication=True, + initdb_params=['--data-checksums'], + pg_options={ + 'max_wal_size': '10GB', + 'max_wal_senders': '2', + 'checkpoint_timeout': '5min', + 'autovacuum': 'off' + } + ) + + self.init_pb(backup_dir) + self.add_instance(backup_dir, 'node', node) + self.set_archiving(backup_dir, 'node', node) + node.slow_start() + + # FULL BACKUP + node.safe_psql( + "postgres", + "create table t_heap as select i as id, md5(i::text) as text, " + "md5(i::text)::tsvector as tsvector from generate_series(0,100) i") + + self.backup_node( + backup_dir, 'node', node) + + # CREATE DATABASE DB1 + node.safe_psql("postgres", "create database db1") + node.safe_psql( + "db1", + "create table t_heap as select i as id, md5(i::text) as text, " + "md5(i::text)::tsvector as tsvector from generate_series(0,1000) i") + + # PAGE BACKUP + backup_id = self.backup_node(backup_dir, 'node', node, backup_type='page') + + if self.paranoia: + pgdata = self.pgdata_content(node.data_dir) + + # RESTORE + node_restored = self.make_simple_node( + base_dir="{0}/{1}/node_restored".format(module_name, fname) + ) + + node_restored.cleanup() + self.restore_node( + backup_dir, 'node', node_restored, + backup_id=backup_id, options=["-j", "4"]) + + # COMPARE PHYSICAL CONTENT + if self.paranoia: + pgdata_restored = self.pgdata_content(node_restored.data_dir) + self.compare_pgdata(pgdata, pgdata_restored) + + # START RESTORED NODE + node_restored.append_conf( + "postgresql.auto.conf", "port = {0}".format(node_restored.port)) + node_restored.slow_start() + + node_restored.safe_psql('db1', 'select 1') + node_restored.cleanup() + + # DROP DATABASE DB1 + node.safe_psql( + "postgres", "drop database db1") + # SECOND PTRACK BACKUP + backup_id = self.backup_node( + backup_dir, 'node', node, backup_type='page') + + if self.paranoia: + pgdata = self.pgdata_content(node.data_dir) + + # RESTORE SECOND PTRACK BACKUP + self.restore_node( + backup_dir, 'node', node_restored, + backup_id=backup_id, options=["-j", "4"] + ) + + # COMPARE PHYSICAL CONTENT + if self.paranoia: + pgdata_restored = self.pgdata_content( + node_restored.data_dir, ignore_ptrack=False) + self.compare_pgdata(pgdata, pgdata_restored) + + # START RESTORED NODE + node_restored.append_conf( + "postgresql.auto.conf", "port = {0}".format(node_restored.port)) + node_restored.slow_start() + + try: + node_restored.safe_psql('db1', 'select 1') + # we should die here because exception is what we expect to happen + self.assertEqual( + 1, 0, + "Expecting Error because we are connecting to deleted database" + "\n Output: {0} \n CMD: {1}".format( + repr(self.output), self.cmd) + ) + except QueryException as e: + self.assertTrue( + 'FATAL: database "db1" does not exist' in e.message, + '\n Unexpected Error Message: {0}\n CMD: {1}'.format( + repr(e.message), self.cmd) + ) + + # Clean after yourself + self.del_test_dir(module_name, fname) From eef5769ffd3da14e9d4faa649b0b162dbd606af6 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Thu, 13 Dec 2018 10:20:49 +0300 Subject: [PATCH 21/23] tests: pgdata_compare now detect missing or redundant directories --- tests/helpers/ptrack_helpers.py | 48 ++++++++++++++++++++++++++++++--- 1 file changed, 44 insertions(+), 4 deletions(-) diff --git a/tests/helpers/ptrack_helpers.py b/tests/helpers/ptrack_helpers.py index d81d7288..0bc27abf 100644 --- a/tests/helpers/ptrack_helpers.py +++ b/tests/helpers/ptrack_helpers.py @@ -1047,7 +1047,7 @@ class ProbackupTest(object): except: pass - def pgdata_content(self, directory, ignore_ptrack=True): + def pgdata_content(self, pgdata, ignore_ptrack=True): """ return dict with directory content. " " TAKE IT AFTER CHECKPOINT or BACKUP""" dirs_to_ignore = [ @@ -1064,9 +1064,10 @@ class ProbackupTest(object): # '_ptrack' # ) directory_dict = {} - directory_dict['pgdata'] = directory + directory_dict['pgdata'] = pgdata directory_dict['files'] = {} - for root, dirs, files in os.walk(directory, followlinks=True): + 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 ( @@ -1076,7 +1077,7 @@ class ProbackupTest(object): continue file_fullpath = os.path.join(root, file) - file_relpath = os.path.relpath(file_fullpath, directory) + file_relpath = os.path.relpath(file_fullpath, pgdata) directory_dict['files'][file_relpath] = {'is_datafile': False} directory_dict['files'][file_relpath]['md5'] = hashlib.md5( open(file_fullpath, 'rb').read()).hexdigest() @@ -1089,12 +1090,51 @@ class ProbackupTest(object): 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'].append(directory_relpath) + return directory_dict def compare_pgdata(self, original_pgdata, restored_pgdata): """ return dict with directory content. DO IT BEFORE RECOVERY""" 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 is not present' + error_message += ' in original PGDATA: {0}\n'.format( + os.path.join(restored_pgdata['pgdata'], directory)) + + 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 From a5cef9890829c90b4df15efee6b5e5cda6c9c782 Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Thu, 13 Dec 2018 11:45:04 +0300 Subject: [PATCH 22/23] tests: test_merge_tablespaces and test_merge_tablespaces_1 added --- tests/helpers/ptrack_helpers.py | 2 +- tests/merge.py | 154 ++++++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+), 1 deletion(-) diff --git a/tests/helpers/ptrack_helpers.py b/tests/helpers/ptrack_helpers.py index 0bc27abf..8ea90c24 100644 --- a/tests/helpers/ptrack_helpers.py +++ b/tests/helpers/ptrack_helpers.py @@ -1123,7 +1123,7 @@ class ProbackupTest(object): for directory in restored_pgdata['dirs']: if directory not in original_pgdata['dirs']: fail = True - error_message += '\nDirectory is not present' + error_message += '\nDirectory was not present' error_message += ' in original PGDATA: {0}\n'.format( os.path.join(restored_pgdata['pgdata'], directory)) diff --git a/tests/merge.py b/tests/merge.py index e4209441..8ce1795e 100644 --- a/tests/merge.py +++ b/tests/merge.py @@ -3,6 +3,7 @@ import unittest import os from .helpers.ptrack_helpers import ProbackupTest, ProbackupException +import shutil module_name = "merge" @@ -471,9 +472,162 @@ class MergeTest(ProbackupTest, unittest.TestCase): # @unittest.skip("skip") def test_merge_tablespaces(self): + """ + Create tablespace with table + """ + fname = self.id().split('.')[3] + backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup') + node = self.make_simple_node( + base_dir="{0}/{1}/node".format(module_name, fname), + set_replication=True, initdb_params=['--data-checksums'], + pg_options={ + 'wal_level': 'replica', + 'max_wal_senders': '2', + 'autovacuum': 'off' + } + ) + + self.init_pb(backup_dir) + self.add_instance(backup_dir, 'node', node) + self.set_archiving(backup_dir, 'node', node) + node.start() + + self.create_tblspace_in_node(node, 'somedata') + node.safe_psql( + "postgres", + "create table t_heap tablespace somedata as select i as id," + " md5(i::text) as text, md5(i::text)::tsvector as tsvector" + " from generate_series(0,100) i" + ) + # FULL backup + self.backup_node(backup_dir, 'node', node) + + # Create new tablespace + self.create_tblspace_in_node(node, 'somedata1') + + node.safe_psql( + "postgres", + "create table t_heap1 tablespace somedata1 as select i as id," + " md5(i::text) as text, md5(i::text)::tsvector as tsvector" + " from generate_series(0,100) i" + ) + + node.safe_psql( + "postgres", + "drop table t_heap" + ) + + # Drop old tablespace + node.safe_psql( + "postgres", + "drop tablespace somedata" + ) + + # PAGE backup + backup_id = self.backup_node(backup_dir, 'node', node, backup_type="page") + + pgdata = self.pgdata_content(node.data_dir) + + node.stop() + shutil.rmtree( + self.get_tblspace_path(node, 'somedata'), + ignore_errors=True) + shutil.rmtree( + self.get_tblspace_path(node, 'somedata1'), + ignore_errors=True) + node.cleanup() + + self.merge_backup(backup_dir, 'node', backup_id) + + self.restore_node( + backup_dir, 'node', node, options=["-j", "4"]) + + pgdata_restored = self.pgdata_content(node.data_dir) + + # this compare should fall because we lost some directories + self.compare_pgdata(pgdata, pgdata_restored) + + # @unittest.skip("skip") + def test_merge_tablespaces_1(self): """ Some test here """ + fname = self.id().split('.')[3] + backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup') + node = self.make_simple_node( + base_dir="{0}/{1}/node".format(module_name, fname), + set_replication=True, initdb_params=['--data-checksums'], + pg_options={ + 'wal_level': 'replica', + 'max_wal_senders': '2', + 'autovacuum': 'off' + } + ) + + self.init_pb(backup_dir) + self.add_instance(backup_dir, 'node', node) + self.set_archiving(backup_dir, 'node', node) + node.start() + + self.create_tblspace_in_node(node, 'somedata') + + # FULL backup + self.backup_node(backup_dir, 'node', node) + node.safe_psql( + "postgres", + "create table t_heap tablespace somedata as select i as id," + " md5(i::text) as text, md5(i::text)::tsvector as tsvector" + " from generate_series(0,100) i" + ) + + # CREATE NEW TABLESPACE + self.create_tblspace_in_node(node, 'somedata1') + + node.safe_psql( + "postgres", + "create table t_heap1 tablespace somedata1 as select i as id," + " md5(i::text) as text, md5(i::text)::tsvector as tsvector" + " from generate_series(0,100) i" + ) + + # PAGE backup + self.backup_node(backup_dir, 'node', node, backup_type="page") + + node.safe_psql( + "postgres", + "drop table t_heap" + ) + node.safe_psql( + "postgres", + "drop tablespace somedata" + ) + + # DELTA backup + backup_id = self.backup_node( + backup_dir, 'node', node, backup_type="delta") + + pgdata = self.pgdata_content(node.data_dir) + + node.stop() + shutil.rmtree( + self.get_tblspace_path(node, 'somedata'), + ignore_errors=True) + shutil.rmtree( + self.get_tblspace_path(node, 'somedata1'), + ignore_errors=True) + node.cleanup() + + self.merge_backup(backup_dir, 'node', backup_id) + + self.restore_node( + backup_dir, 'node', node, + options=["-j", "4"]) + + pgdata_restored = self.pgdata_content(node.data_dir) + self.compare_pgdata(pgdata, pgdata_restored) + + # Clean after yourself + self.del_test_dir(module_name, fname) def test_merge_page_truncate(self): """ From e16bdc6660bb5666c485a897e4f444bcdf3d9aab Mon Sep 17 00:00:00 2001 From: Grigory Smolkin Date: Thu, 13 Dec 2018 13:21:57 +0300 Subject: [PATCH 23/23] tests: add some comments --- tests/merge.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/tests/merge.py b/tests/merge.py index 8ce1795e..2dbe8ee0 100644 --- a/tests/merge.py +++ b/tests/merge.py @@ -473,7 +473,10 @@ class MergeTest(ProbackupTest, unittest.TestCase): # @unittest.skip("skip") def test_merge_tablespaces(self): """ - Create tablespace with table + Create tablespace with table, take FULL backup, + create another tablespace with another table and drop previous + tablespace, take page backup, merge it and restore + """ fname = self.id().split('.')[3] backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup') @@ -550,7 +553,10 @@ class MergeTest(ProbackupTest, unittest.TestCase): # @unittest.skip("skip") def test_merge_tablespaces_1(self): """ - Some test here + Create tablespace with table, take FULL backup, + create another tablespace with another table, take page backup, + drop first tablespace and take delta backup, + merge it and restore """ fname = self.id().split('.')[3] backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup') @@ -1202,3 +1208,4 @@ class MergeTest(ProbackupTest, unittest.TestCase): # FULL MERGING # 3. Need new test with corrupted FULL backup +# 4. different compression levels