db/pg_probackup
1
0
Fork 0
mirror of https://github.com/postgrespro/pg_probackup.git synced 2026-06-21 01:34:15 +02:00
Code Issues Releases Activity

[Issue #185] archive-get refactoring: multi-threading, batching, prefetchin, retrying and handling of '.partial' files

Browse Source
This commit is contained in:
Grigory Smolkin
2020-04-17 03:40:23 +03:00
parent 439c27c7e7
commit 703d311b4b
11 changed files with 1658 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/pgprobackup.xml
+45 -6
View File
@@ -3957,7 +3957,7 @@ pg_probackup delete -B <replaceable>backup_dir</replaceable> --instance <replace
<title>archive-push</title>
<programlisting>
pg_probackup archive-push -B <replaceable>backup_dir</replaceable> --instance <replaceable>instance_name</replaceable>
--wal-file-name=<replaceable>wal_file_name</replaceable>
--wal-file-name=<replaceable>wal_file_name</replaceable> [--wal-file-path=<replaceable>wal_file_path</replaceable>]
[--help] [--no-sync] [--compress] [--no-ready-rename] [--overwrite]
[-j <replaceable>num_threads</replaceable>] [--batch-size=<replaceable>batch_size</replaceable>]
[--archive-timeout=<replaceable>timeout</replaceable>]
@@ -3973,12 +3973,10 @@ pg_probackup archive-push -B <replaceable>backup_dir</replaceable> --instance <r
backup instance and the cluster do not match, this command
fails with the following error message: <literal>Refuse to push WAL
segment segment_name into archive. Instance parameters
mismatch.</literal> For each WAL file moved to the backup catalog, you
will see the following message in the <productname>PostgreSQL</productname> log file:
<literal>pg_probackup archive-push completed successfully</literal>.
mismatch.</literal>
</para>
<para>
If the files to be copied already exist in the backup catalog,
If the files to be copied already exists in the backup catalog,
<application>pg_probackup</application> computes and compares their checksums. If the
checksums match, <command>archive-push</command> skips the corresponding file and
returns a successful execution code. Otherwise, <command>archive-push</command>
@@ -4025,6 +4023,8 @@ pg_probackup archive-push -B <replaceable>backup_dir</replaceable> --instance <r
<title>archive-get</title>
<programlisting>
pg_probackup archive-get -B <replaceable>backup_dir</replaceable> --instance <replaceable>instance_name</replaceable> --wal-file-path=<replaceable>wal_file_path</replaceable> --wal-file-name=<replaceable>wal_file_name</replaceable>
[-j <replaceable>num_threads</replaceable>] [--batch-size=<replaceable>batch_size</replaceable>]
[--prefetch-dir=<replaceable>prefetch_dir_path</replaceable>] [--no-validate-wal]
[--help] [<replaceable>remote_options</replaceable>] [<replaceable>logging_options</replaceable>]
</programlisting>
<para>
@@ -4035,6 +4035,17 @@ pg_probackup archive-get -B <replaceable>backup_dir</replaceable> --instance <re
restoring backups using a WAL archive. You do not need to set
it manually.
</para>
<para>
To speed up recovery, you can specify the <option>-j</option> option
to run <command>archive-get</command> on multiple threads.
If you provide the <option>--batch-size</option> option, WAL segments
will be copied in batches of the specified size.
</para>
<para>
For details, see section <link linkend="pbk-archiving-options">Archiving Options</link>.
</para>
</refsect3>
</refsect2>
<refsect2 id="pbk-options">
@@ -4777,7 +4788,8 @@ pg_probackup archive-get -B <replaceable>backup_dir</replaceable> --instance <re
<listitem>
<para>
Sets the maximum number of files that can be copied into the archive
by a single <command>archive-push</command> process.
by a single <command>archive-push</command> process, or from
the archive by a single <command>archive-get</command> process.
</para>
</listitem>
</varlistentry>
@@ -4789,6 +4801,7 @@ pg_probackup archive-get -B <replaceable>backup_dir</replaceable> --instance <re
Sets the timeout for considering existing <literal>.part</literal>
files to be stale. By default, <application>pg_probackup</application>
waits 300 seconds.
This option can be used only with <xref linkend="pbk-archive-push"/> command.
</para>
</listitem>
</varlistentry>
@@ -4800,6 +4813,7 @@ pg_probackup archive-get -B <replaceable>backup_dir</replaceable> --instance <re
Do not rename status files in the <literal>archive_status</literal> directory.
This option should be used only if <parameter>archive_command</parameter>
contains multiple commands.
This option can be used only with <xref linkend="pbk-archive-push"/> command.
</para>
</listitem>
</varlistentry>
@@ -4811,6 +4825,31 @@ pg_probackup archive-get -B <replaceable>backup_dir</replaceable> --instance <re
Do not sync copied WAL files to disk. You can use this flag to speed
up archiving process. Using this flag can result in WAL archive
corruption in case of operating system or hardware crash.
This option can be used only with <xref linkend="pbk-archive-push"/> command.
</para>
</listitem>
</varlistentry>
<varlistentry>
<term><option>--prefetch-dir=<replaceable>path</replaceable></option></term>
<listitem>
<para>
Directory used to store prefetched WAL segments if <option>--batch-size</option> option is used.
Directory must be located on the same filesystem and on the same mountpoint the
<literal>PGDATA/pg_wal</literal> is located.
By default files are stored in <literal>PGDATA/pg_wal/pbk_prefetch</literal> directory.
This option can be used only with <xref linkend="pbk-archive-get"/> command.
</para>
</listitem>
</varlistentry>
<varlistentry>
<term><option>--no-validate-wal</option></term>
<listitem>
<para>
Do not validate prefetched WAL file before using it.
Use this option if you want to increase the speed of recovery.
This option can be used only with <xref linkend="pbk-archive-get"/> command.
</para>
</listitem>
</varlistentry>
src/archive.c
+815 -240
View File
File diff suppressed because it is too large Load Diff
src/backup.c
+1 -1
View File
@@ -443,7 +443,7 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
/* TODO: add ms precision */
if (pagemap_isok)
elog(INFO, "Pagemap successfully extracted, time elapsed %.0f sec",
elog(INFO, "Pagemap successfully extracted, time elapsed: %.0f sec",
difftime(end_time, start_time));
else
elog(ERROR, "Pagemap extraction failed, time elasped: %.0f sec",
src/help.c
+10 -2
View File
@@ -215,7 +215,7 @@ help_pg_probackup(void)
printf(_("\n %s archive-push -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" [-j num-threads] [--batch-size batch_size]\n"));
printf(_(" [-j num-threads] [--batch-size=batch_size]\n"));
printf(_(" [--archive-timeout=timeout]\n"));
printf(_(" [--no-ready-rename] [--no-sync]\n"));
printf(_(" [--overwrite] [--compress]\n"));
@@ -229,6 +229,8 @@ help_pg_probackup(void)
printf(_("\n %s archive-get -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" --wal-file-path=wal-file-path\n"));
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" [-j num-threads] [--batch-size=batch_size]\n"));
printf(_(" [--no-validate-wal]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
@@ -870,7 +872,7 @@ help_archive_push(void)
{
printf(_("\n%s archive-push -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" [-j num-threads] [--batch-size batch_size]\n"));
printf(_(" [-j num-threads] [--batch-size=batch_size]\n"));
printf(_(" [--archive-timeout=timeout]\n"));
printf(_(" [--no-ready-rename] [--no-sync]\n"));
printf(_(" [--overwrite] [--compress]\n"));
@@ -916,6 +918,8 @@ help_archive_get(void)
printf(_("\n%s archive-get -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" --wal-file-path=wal-file-path\n"));
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" [-j num-threads] [--batch-size=batch_size]\n"));
printf(_(" [--no-validate-wal]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n\n"));
@@ -926,6 +930,10 @@ help_archive_get(void)
printf(_(" relative destination path name of the WAL file on the server\n"));
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" name of the WAL file to retrieve from the archive\n"));
printf(_(" -j, --threads=NUM number of parallel threads\n"));
printf(_(" --batch-size=NUM number of files to be prefetched\n"));
printf(_(" --prefetch-dir=path location of the store area for prefetched WAL files\n"));
printf(_(" --no-validate-wal skip validation of prefetched WAL file before using it\n"));
printf(_("\n Remote options:\n"));
printf(_(" --remote-proto=protocol remote protocol to use\n"));
src/parsexlog.c
+38 -12
View File
@@ -325,12 +325,15 @@ extractPageMap(const char *archivedir, uint32 wal_seg_size,
for (i = parray_num(interval_list) - 1; i >= 0; i--)
{
bool inclusive_endpoint = false;
bool inclusive_endpoint;
wal_interval = parray_get(interval_list, i);
/* In case of replica promotion, endpoints of intermediate
* timelines can be unreachable.
*/
inclusive_endpoint = false;
/* ... but not the end timeline */
if (wal_interval->tli == end_tli)
inclusive_endpoint = true;
@@ -453,7 +456,7 @@ validate_wal(pgBackup *backup, const char *archivedir,
* If recovery target is provided, ensure that archive files exist in
* archive directory.
*/
if (dir_is_empty(archivedir, FIO_BACKUP_HOST))
if (dir_is_empty(archivedir, FIO_LOCAL_HOST))
elog(ERROR, "WAL archive is empty. You cannot restore backup to a recovery target without WAL archive.");
/*
@@ -872,20 +875,20 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
/* If segment do not exists, but the same
* segment with '.partial' suffix does, use it instead */
if (!fileExists(reader_data->xlogpath, FIO_BACKUP_HOST) &&
fileExists(partial_file, FIO_BACKUP_HOST))
if (!fileExists(reader_data->xlogpath, FIO_LOCAL_HOST) &&
fileExists(partial_file, FIO_LOCAL_HOST))
{
snprintf(reader_data->xlogpath, MAXPGPATH, "%s", partial_file);
}
if (fileExists(reader_data->xlogpath, FIO_BACKUP_HOST))
if (fileExists(reader_data->xlogpath, FIO_LOCAL_HOST))
{
elog(LOG, "Thread [%d]: Opening WAL segment \"%s\"",
reader_data->thread_num, reader_data->xlogpath);
reader_data->xlogexists = true;
reader_data->xlogfile = fio_open(reader_data->xlogpath,
O_RDONLY | PG_BINARY, FIO_BACKUP_HOST);
O_RDONLY | PG_BINARY, FIO_LOCAL_HOST);
if (reader_data->xlogfile < 0)
{
@@ -897,14 +900,14 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
}
#ifdef HAVE_LIBZ
/* Try to open compressed WAL segment */
else if (fileExists(reader_data->gz_xlogpath, FIO_BACKUP_HOST))
else if (fileExists(reader_data->gz_xlogpath, FIO_LOCAL_HOST))
{
elog(LOG, "Thread [%d]: Opening compressed WAL segment \"%s\"",
reader_data->thread_num, reader_data->gz_xlogpath);
reader_data->xlogexists = true;
reader_data->gz_xlogfile = fio_gzopen(reader_data->gz_xlogpath,
"rb", -1, FIO_BACKUP_HOST);
"rb", -1, FIO_LOCAL_HOST);
if (reader_data->gz_xlogfile == NULL)
{
elog(WARNING, "Thread [%d]: Could not open compressed WAL segment \"%s\": %s",
@@ -1045,15 +1048,13 @@ RunXLogThreads(const char *archivedir, time_t target_time,
XLogSegNo endSegNo = 0;
bool result = true;
if (!XRecOffIsValid(startpoint) &&
!XRecOffIsNull(startpoint))
{
if (!XRecOffIsValid(startpoint) && !XRecOffIsNull(startpoint))
elog(ERROR, "Invalid startpoint value %X/%X",
(uint32) (startpoint >> 32), (uint32) (startpoint));
}
if (!XLogRecPtrIsInvalid(endpoint))
{
// if (XRecOffIsNull(endpoint) && !inclusive_endpoint)
if (XRecOffIsNull(endpoint))
{
GetXLogSegNo(endpoint, endSegNo, segment_size);
@@ -1719,3 +1720,28 @@ getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
return false;
}
bool validate_wal_segment(TimeLineID tli, XLogSegNo segno, const char *prefetch_dir, uint32 wal_seg_size)
{
XLogRecPtr startpoint;
XLogRecPtr endpoint;
bool rc;
int tmp_num_threads = num_threads;
num_threads = 1;
/* calculate startpoint and endpoint */
GetXLogRecPtr(segno, 0, wal_seg_size, startpoint);
GetXLogRecPtr(segno+1, 0, wal_seg_size, endpoint);
/* disable multi-threading */
num_threads = 1;
rc = RunXLogThreads(prefetch_dir, 0, InvalidTransactionId,
InvalidXLogRecPtr, tli, wal_seg_size,
startpoint, endpoint, false, NULL, NULL, true);
num_threads = tmp_num_threads;
return rc;
}
src/pg_probackup.c
+28 -17
View File
@@ -131,6 +131,10 @@ static char *wal_file_name;
static bool file_overwrite = false;
static bool no_ready_rename = false;
/* archive get options */
static char *prefetch_dir;
bool no_validate_wal = false;
/* show options */
ShowFormat show_format = SHOW_PLAIN;
bool show_archive = false;
@@ -220,9 +224,12 @@ static ConfigOption cmd_options[] =
{ 'b', 152, "overwrite", &file_overwrite, SOURCE_CMD_STRICT },
{ 'b', 153, "no-ready-rename", &no_ready_rename, SOURCE_CMD_STRICT },
{ 'i', 162, "batch-size", &batch_size, SOURCE_CMD_STRICT },
/* archive-get options */
{ 's', 163, "prefetch-dir", &prefetch_dir, SOURCE_CMD_STRICT },
{ 'b', 164, "no-validate-wal", &no_validate_wal, SOURCE_CMD_STRICT },
/* show options */
{ 'f', 163, "format", opt_show_format, SOURCE_CMD_STRICT },
{ 'b', 164, "archive", &show_archive, SOURCE_CMD_STRICT },
{ 'f', 165, "format", opt_show_format, SOURCE_CMD_STRICT },
{ 'b', 166, "archive", &show_archive, SOURCE_CMD_STRICT },
/* set-backup options */
{ 'I', 170, "ttl", &ttl, SOURCE_CMD_STRICT, SOURCE_DEFAULT, 0, OPTION_UNIT_S, option_get_value},
{ 's', 171, "expire-time", &expire_time_string, SOURCE_CMD_STRICT },
@@ -266,9 +273,6 @@ main(int argc, char *argv[])
{
char *command = NULL,
*command_name;
/* Check if backup_path is directory. */
struct stat stat_buf;
int rc;
PROGRAM_NAME_FULL = argv[0];
@@ -448,11 +452,6 @@ main(int argc, char *argv[])
/* Ensure that backup_path is an absolute path */
if (!is_absolute_path(backup_path))
elog(ERROR, "-B, --backup-path must be an absolute path");
/* Ensure that backup_path is a path to a directory */
rc = stat(backup_path, &stat_buf);
if (rc != -1 && !S_ISDIR(stat_buf.st_mode))
elog(ERROR, "-B, --backup-path must be a path to directory");
}
/* Ensure that backup_path is an absolute path */
@@ -504,12 +503,16 @@ main(int argc, char *argv[])
/*
* Ensure that requested backup instance exists.
* for all commands except init, which doesn't take this parameter
* and add-instance which creates new instance.
* for all commands except init, which doesn't take this parameter,
* add-instance which creates new instance
* and archive-get, which just do not require it at this point
*/
if (backup_subcmd != INIT_CMD && backup_subcmd != ADD_INSTANCE_CMD)
if (backup_subcmd != INIT_CMD && backup_subcmd != ADD_INSTANCE_CMD &&
backup_subcmd != ARCHIVE_GET_CMD)
{
if (fio_access(backup_instance_path, F_OK, FIO_BACKUP_HOST) != 0)
struct stat st;
if (fio_stat(backup_instance_path, &st, true, FIO_BACKUP_HOST) != 0)
{
elog(WARNING, "Failed to access directory \"%s\": %s",
backup_instance_path, strerror(errno));
@@ -518,6 +521,12 @@ main(int argc, char *argv[])
elog(ERROR, "Instance '%s' does not exist in this backup catalog",
instance_name);
}
else
{
/* Ensure that backup_path is a path to a directory */
if (!S_ISDIR(st.st_mode))
elog(ERROR, "-B, --backup-path must be a path to directory");
}
}
}
@@ -533,7 +542,8 @@ main(int argc, char *argv[])
config_get_opt_env(instance_options);
/* Read options from configuration file */
if (backup_subcmd != ADD_INSTANCE_CMD)
if (backup_subcmd != ADD_INSTANCE_CMD &&
backup_subcmd != ARCHIVE_GET_CMD)
{
join_path_components(path, backup_instance_path,
BACKUP_CATALOG_CONF_FILE);
@@ -763,8 +773,9 @@ main(int argc, char *argv[])
batch_size, file_overwrite, no_sync, no_ready_rename);
break;
case ARCHIVE_GET_CMD:
return do_archive_get(&instance_config,
wal_file_path, wal_file_name);
do_archive_get(&instance_config, prefetch_dir,
wal_file_path, wal_file_name, batch_size, !no_validate_wal);
break;
case ADD_INSTANCE_CMD:
return do_add_instance(&instance_config);
case DELETE_INSTANCE_CMD:
src/pg_probackup.h
+20 -7
View File
@@ -707,8 +707,8 @@ extern int do_add_instance(InstanceConfig *instance);
extern void do_archive_push(InstanceConfig *instance, char *wal_file_path,
char *wal_file_name, int batch_size, bool overwrite,
bool no_sync, bool no_ready_rename);
extern int do_archive_get(InstanceConfig *instance, char *wal_file_path,
char *wal_file_name);
extern void do_archive_get(InstanceConfig *instance, const char *prefetch_dir_arg, char *wal_file_path,
char *wal_file_name, int batch_size, bool validate_wal);
/* in configure.c */
extern void do_show_config(void);
@@ -922,6 +922,8 @@ extern void validate_wal(pgBackup *backup, const char *archivedir,
time_t target_time, TransactionId target_xid,
XLogRecPtr target_lsn, TimeLineID tli,
uint32 seg_size);
extern bool validate_wal_segment(TimeLineID tli, XLogSegNo segno,
const char *prefetch_dir, uint32 wal_seg_size);
extern bool read_recovery_info(const char *archivedir, TimeLineID tli,
uint32 seg_size,
XLogRecPtr start_lsn, XLogRecPtr stop_lsn,
@@ -990,12 +992,23 @@ extern parray * pg_ptrack_get_pagemapset(PGconn *backup_conn, const char *ptrack
extern int fio_send_pages(FILE* in, FILE* out, pgFile *file, XLogRecPtr horizonLsn,
int calg, int clevel, uint32 checksum_version,
datapagemap_t *pagemap, BlockNumber* err_blknum, char **errormsg);
/* return codes for fio_send_pages */
#define WRITE_FAILED (-1)
#define REMOTE_ERROR (-2)
#define PAGE_CORRUPTION (-3)
#define SEND_OK (-4)
#define OUT_BUF_SIZE (1024 * 1024)
extern int fio_send_file_gz(const char *from_fullpath, const char *to_fullpath, FILE* out, int thread_num);
extern int fio_send_file(const char *from_fullpath, const char *to_fullpath, FILE* out, int thread_num);
/* return codes for fio_send_pages() and fio_send_file() */
#define SEND_OK (0)
#define FILE_MISSING (-1)
#define OPEN_FAILED (-2)
#define READ_FAILED (-3)
#define WRITE_FAILED (-4)
#define ZLIB_ERROR (-5)
#define REMOTE_ERROR (-6)
#define PAGE_CORRUPTION (-8)
/* Check if specified location is local for current node */
extern bool fio_is_remote(fio_location location);
extern void get_header_errormsg(Page page, char **errormsg);
extern void get_checksum_errormsg(Page page, char **errormsg,
src/utils/file.c
+342 -6
View File
@@ -14,6 +14,7 @@
#define PRINTF_BUF_SIZE 1024
#define FILE_PERMISSIONS 0600
#define CHUNK_SIZE 1024 * 128
static __thread unsigned long fio_fdset = 0;
static __thread void* fio_stdin_buffer;
@@ -136,7 +137,7 @@ static int remove_file_or_dir(char const* path)
#endif
/* Check if specified location is local for current node */
static bool fio_is_remote(fio_location location)
bool fio_is_remote(fio_location location)
{
bool is_remote = MyLocation != FIO_LOCAL_HOST
&& location != FIO_LOCAL_HOST
@@ -1026,7 +1027,6 @@ int fio_chmod(char const* path, int mode, fio_location location)
#ifdef HAVE_LIBZ
#define ZLIB_BUFFER_SIZE (64*1024)
#define MAX_WBITS 15 /* 32K LZ77 window */
#define DEF_MEM_LEVEL 8
@@ -1304,8 +1304,10 @@ z_off_t fio_gzseek(gzFile f, z_off_t offset, int whence)
#endif
/* Send file content */
static void fio_send_file(int out, char const* path)
/* Send file content
* Note: it should not be used for large files.
*/
static void fio_load_file(int out, char const* path)
{
int fd = open(path, O_RDONLY);
fio_header hdr;
@@ -1462,7 +1464,7 @@ int fio_send_pages(FILE* in, FILE* out, pgFile *file, XLogRecPtr horizonLsn,
file->uncompressed_size += BLCKSZ;
}
else
elog(ERROR, "Remote agent returned message of unknown type");
elog(ERROR, "Remote agent returned message of unexpected type: %i", hdr.cop);
}
return n_blocks_read;
@@ -1629,6 +1631,337 @@ cleanup:
return;
}
/* Receive chunks of compressed data, decompress them and write to
* destination file.
* Return codes:
* FILE_MISSING (-1)
* OPEN_FAILED (-2)
* READ_FAILED (-3)
* WRITE_FAILED (-4)
* ZLIB_ERROR (-5)
* REMOTE_ERROR (-6)
*/
int fio_send_file_gz(const char *from_fullpath, const char *to_fullpath, FILE* out, int thread_num)
{
fio_header hdr;
int exit_code = SEND_OK;
char *in_buf = pgut_malloc(CHUNK_SIZE); /* buffer for compressed data */
char *out_buf = pgut_malloc(OUT_BUF_SIZE); /* 1MB buffer for decompressed data */
size_t path_len = strlen(from_fullpath) + 1;
/* decompressor */
z_stream *strm = NULL;
hdr.cop = FIO_SEND_FILE;
hdr.size = path_len;
elog(VERBOSE, "Thread [%d]: Attempting to open remote compressed WAL file '%s'",
thread_num, from_fullpath);
IO_CHECK(fio_write_all(fio_stdout, &hdr, sizeof(hdr)), sizeof(hdr));
IO_CHECK(fio_write_all(fio_stdout, from_fullpath, path_len), path_len);
for (;;)
{
fio_header hdr;
IO_CHECK(fio_read_all(fio_stdin, &hdr, sizeof(hdr)), sizeof(hdr));
if (hdr.cop == FIO_SEND_FILE_EOF)
{
break;
}
else if (hdr.cop == FIO_ERROR)
{
/* handle error, reported by the agent */
if (hdr.size > 0)
{
IO_CHECK(fio_read_all(fio_stdin, in_buf, hdr.size), hdr.size);
elog(WARNING, "Thread [%d]: %s", thread_num, in_buf);
}
exit_code = hdr.arg;
goto cleanup;
}
else if (hdr.cop == FIO_PAGE)
{
int rc;
Assert(hdr.size <= CHUNK_SIZE);
IO_CHECK(fio_read_all(fio_stdin, in_buf, hdr.size), hdr.size);
/* We have received a chunk of compressed data, lets decompress it */
if (strm == NULL)
{
/* Initialize decompressor */
strm = pgut_malloc(sizeof(z_stream));
memset(strm, 0, sizeof(z_stream));
/* The fields next_in, avail_in initialized before init */
strm->next_in = (Bytef *)in_buf;
strm->avail_in = hdr.size;
rc = inflateInit2(strm, 15 + 16);
if (rc != Z_OK)
{
elog(WARNING, "Thread [%d]: Failed to initialize decompression stream for file '%s': %i: %s",
thread_num, from_fullpath, rc, strm->msg);
exit_code = ZLIB_ERROR;
goto cleanup;
}
}
else
{
strm->next_in = (Bytef *)in_buf;
strm->avail_in = hdr.size;
}
strm->next_out = (Bytef *)out_buf; /* output buffer */
strm->avail_out = OUT_BUF_SIZE; /* free space in output buffer */
/*
* From zlib documentation:
* The application must update next_in and avail_in when avail_in
* has dropped to zero. It must update next_out and avail_out when
* avail_out has dropped to zero.
*/
while (strm->avail_in != 0) /* while there is data in input buffer, decompress it */
{
/* decompress until there is no data to decompress,
* or buffer with uncompressed data is full
*/
rc = inflate(strm, Z_NO_FLUSH);
if (rc == Z_STREAM_END)
/* end of stream */
break;
else if (rc != Z_OK)
{
/* got an error */
elog(WARNING, "Thread [%d]: Decompression failed for file '%s': %i: %s",
thread_num, from_fullpath, rc, strm->msg);
exit_code = ZLIB_ERROR;
goto cleanup;
}
if (strm->avail_out == 0)
{
/* Output buffer is full, write it out */
if (fwrite(out_buf, 1, OUT_BUF_SIZE, out) != OUT_BUF_SIZE)
{
elog(WARNING, "Thread [%d]: Cannot write to file '%s': %s",
thread_num, to_fullpath, strerror(errno));
exit_code = WRITE_FAILED;
goto cleanup;
}
strm->next_out = (Bytef *)out_buf; /* output buffer */
strm->avail_out = OUT_BUF_SIZE;
}
}
/* write out leftovers if any */
if (strm->avail_out != OUT_BUF_SIZE)
{
int len = OUT_BUF_SIZE - strm->avail_out;
if (fwrite(out_buf, 1, len, out) != len)
{
elog(WARNING, "Thread [%d]: Cannot write to file: %s",
thread_num, strerror(errno));
exit_code = WRITE_FAILED;
goto cleanup;
}
}
}
else
{
elog(WARNING, "Thread [%d]: Remote agent returned message of unexpected type: %i",
thread_num, hdr.cop);
exit_code = REMOTE_ERROR;
break;
}
}
cleanup:
if (exit_code < OPEN_FAILED)
fio_disconnect(); /* discard possible pending data in pipe */
if (strm)
{
inflateEnd(strm);
pg_free(strm);
}
pg_free(in_buf);
pg_free(out_buf);
return exit_code;
}
/* Receive chunks of data and write them to destination file.
* Return codes:
* SEND_OK (0)
* FILE_MISSING (-1)
* OPEN_FAILED (-2)
* READ_FAIL (-3)
* WRITE_FAIL (-4)
*/
int fio_send_file(const char *from_fullpath, const char *to_fullpath, FILE* out, int thread_num)
{
fio_header hdr;
int exit_code = SEND_OK;
size_t path_len = strlen(from_fullpath) + 1;
char *buf = pgut_malloc(CHUNK_SIZE); /* buffer */
hdr.cop = FIO_SEND_FILE;
hdr.size = path_len;
elog(VERBOSE, "Thread [%d]: Attempting to open remote WAL file '%s'",
thread_num, from_fullpath);
IO_CHECK(fio_write_all(fio_stdout, &hdr, sizeof(hdr)), sizeof(hdr));
IO_CHECK(fio_write_all(fio_stdout, from_fullpath, path_len), path_len);
for (;;)
{
/* receive data */
IO_CHECK(fio_read_all(fio_stdin, &hdr, sizeof(hdr)), sizeof(hdr));
if (hdr.cop == FIO_SEND_FILE_EOF)
{
break;
}
else if (hdr.cop == FIO_ERROR)
{
/* handle error, reported by the agent */
if (hdr.size > 0)
{
IO_CHECK(fio_read_all(fio_stdin, buf, hdr.size), hdr.size);
elog(WARNING, "Thread [%d]: %s", thread_num, buf);
}
exit_code = hdr.arg;
break;
}
else if (hdr.cop == FIO_PAGE)
{
Assert(hdr.size <= CHUNK_SIZE);
IO_CHECK(fio_read_all(fio_stdin, buf, hdr.size), hdr.size);
/* We have received a chunk of data data, lets write it out */
if (fwrite(buf, 1, hdr.size, out) != hdr.size)
{
elog(WARNING, "Thread [%d]: Cannot write to file '%s': %s",
thread_num, to_fullpath, strerror(errno));
exit_code = WRITE_FAILED;
break;
}
}
else
{
/* TODO: fio_disconnect may get assert fail when running after this */
elog(WARNING, "Thread [%d]: Remote agent returned message of unexpected type: %i",
thread_num, hdr.cop);
exit_code = REMOTE_ERROR;
break;
}
}
if (exit_code < OPEN_FAILED)
fio_disconnect(); /* discard possible pending data in pipe */
pg_free(buf);
return exit_code;
}
/* Send file content
* On error we return FIO_ERROR message with following codes
* FILE_MISSING (-1)
* OPEN_FAILED (-2)
* READ_FAILED (-3)
*
*/
static void fio_send_file_impl(int out, char const* path)
{
FILE *fp;
fio_header hdr;
char *buf = pgut_malloc(CHUNK_SIZE);
ssize_t read_len = 0;
char *errormsg = NULL;
/* open source file for read */
/* TODO: check that file is regular file */
fp = fopen(path, PG_BINARY_R);
if (!fp)
{
hdr.cop = FIO_ERROR;
/* do not send exact wording of ENOENT error message
* because it is a very common error in our case, so
* error code is enough.
*/
if (errno == ENOENT)
{
hdr.arg = FILE_MISSING;
hdr.size = 0;
}
else
{
hdr.arg = OPEN_FAILED;
errormsg = pgut_malloc(MAXPGPATH);
/* Construct the error message */
snprintf(errormsg, MAXPGPATH, "Cannot open source file '%s': %s", path, strerror(errno));
hdr.size = strlen(errormsg) + 1;
}
/* send header and message */
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
if (errormsg)
IO_CHECK(fio_write_all(out, errormsg, hdr.size), hdr.size);
goto cleanup;
}
/* copy content */
for (;;)
{
read_len = fread(buf, 1, CHUNK_SIZE, fp);
/* report error */
if (read_len < 0 || (read_len == 0 && !feof(fp)))
{
hdr.cop = FIO_ERROR;
errormsg = pgut_malloc(MAXPGPATH);
hdr.arg = READ_FAILED;
/* Construct the error message */
snprintf(errormsg, MAXPGPATH, "Cannot read source file '%s': %s", path, strerror(errno));
hdr.size = strlen(errormsg) + 1;
/* send header and message */
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
IO_CHECK(fio_write_all(out, errormsg, hdr.size), hdr.size);
goto cleanup;
}
else if (read_len == 0)
break;
else
{
/* send chunk */
hdr.cop = FIO_PAGE;
hdr.size = read_len;
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
IO_CHECK(fio_write_all(out, buf, read_len), read_len);
}
}
/* we are done, send eof */
hdr.cop = FIO_SEND_FILE_EOF;
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
cleanup:
if (fp)
fclose(fp);
pg_free(buf);
pg_free(errormsg);
return;
}
/* Execute commands at remote host */
void fio_communicate(int in, int out)
{
@@ -1665,7 +1998,7 @@ void fio_communicate(int in, int out)
}
switch (hdr.cop) {
case FIO_LOAD: /* Send file content */
fio_send_file(out, buf);
fio_load_file(out, buf);
break;
case FIO_OPENDIR: /* Open directory for traversal */
dir[hdr.handle] = opendir(buf);
@@ -1776,6 +2109,9 @@ void fio_communicate(int in, int out)
// buf contain fio_send_request header and bitmap.
fio_send_pages_impl(fd[hdr.handle], out, buf, true);
break;
case FIO_SEND_FILE:
fio_send_file_impl(out, buf);
break;
case FIO_SYNC:
/* open file and fsync it */
tmp_fd = open(buf, O_WRONLY | PG_BINARY, FILE_PERMISSIONS);
src/utils/file.h
+2
View File
@@ -40,6 +40,8 @@ typedef enum
FIO_SEND_PAGES,
FIO_SEND_PAGES_PAGEMAP,
FIO_ERROR,
FIO_SEND_FILE,
// FIO_CHUNK,
FIO_SEND_FILE_EOF,
FIO_SEND_FILE_CORRUPTION,
/* messages for closing connection */
tests/archive.py
+340 -4
View File
@@ -398,7 +398,7 @@ class ArchiveTest(ProbackupTest, unittest.TestCase):
log_content)
self.assertIn(
'pg_probackup push file',
'pg_probackup archive-push WAL file',
log_content)
self.assertIn(
@@ -482,7 +482,7 @@ class ArchiveTest(ProbackupTest, unittest.TestCase):
self.assertIn(
'DETAIL: The failed archive command was:', log_content)
self.assertIn(
'pg_probackup push file', log_content)
'pg_probackup archive-push WAL file', log_content)
self.assertNotIn(
'WAL file already exists in archive with '
'different checksum, overwriting', log_content)
@@ -941,11 +941,11 @@ class ArchiveTest(ProbackupTest, unittest.TestCase):
pgdata_replica = self.pgdata_content(replica.data_dir)
self.compare_pgdata(pgdata_master, pgdata_replica)
self.set_replica(master, replica, synchronous=True)
self.set_replica(master, replica, synchronous=False)
# ADD INSTANCE REPLICA
# self.add_instance(backup_dir, 'replica', replica)
# SET ARCHIVING FOR REPLICA
# self.set_archiving(backup_dir, 'replica', replica, replica=True)
self.set_archiving(backup_dir, 'master', replica, replica=True)
replica.slow_start(replica=True)
# CHECK LOGICAL CORRECTNESS on REPLICA
@@ -977,6 +977,18 @@ class ArchiveTest(ProbackupTest, unittest.TestCase):
self.assertEqual(
'OK', self.show_pb(backup_dir, 'master', backup_id)['status'])
master.pgbench_init(scale=50)
sleep(10)
replica.promote()
master.pgbench_init(scale=10)
replica.pgbench_init(scale=10)
exit(1)
# Clean after yourself
self.del_test_dir(module_name, fname)
@@ -1984,6 +1996,11 @@ class ArchiveTest(ProbackupTest, unittest.TestCase):
self.set_auto_conf(node_restored, {'port': node_restored.port})
self.set_auto_conf(node_restored, {'hot_standby': 'off'})
# it will set node_restored as warm standby.
# with open(os.path.join(node_restored.data_dir, "standby.signal"), 'w') as f:
# f.flush()
# f.close()
node_restored.slow_start()
result = node.safe_psql(
@@ -2000,7 +2017,326 @@ class ArchiveTest(ProbackupTest, unittest.TestCase):
pg_receivexlog.kill()
self.del_test_dir(module_name, fname)
def test_multi_timeline_recovery_prefetching(self):
""""""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
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()
self.backup_node(backup_dir, 'node', node)
node.pgbench_init(scale=50)
target_xid = node.safe_psql(
'postgres',
'select txid_current()').rstrip()
node.pgbench_init(scale=20)
node.stop()
node.cleanup()
self.restore_node(
backup_dir, 'node', node,
options=[
'--recovery-target-xid={0}'.format(target_xid),
'--recovery-target-action=promote'])
node.slow_start()
node.pgbench_init(scale=20)
target_xid = node.safe_psql(
'postgres',
'select txid_current()').rstrip()
node.stop(['-m', 'immediate', '-D', node.data_dir])
node.cleanup()
self.restore_node(
backup_dir, 'node', node,
options=[
# '--recovery-target-xid={0}'.format(target_xid),
'--recovery-target-timeline=2',
# '--recovery-target-action=promote',
'--no-validate'])
node.slow_start()
node.pgbench_init(scale=20)
result = node.safe_psql(
'postgres',
'select * from pgbench_accounts')
node.stop()
node.cleanup()
self.restore_node(
backup_dir, 'node', node,
options=[
# '--recovery-target-xid=100500',
'--recovery-target-timeline=3',
# '--recovery-target-action=promote',
'--no-validate'])
os.remove(os.path.join(node.logs_dir, 'postgresql.log'))
restore_command = self.get_restore_command(backup_dir, 'node', node)
restore_command += ' -j 2 --batch-size=10 --log-level-console=VERBOSE'
if node.major_version >= 12:
node.append_conf(
'probackup_recovery.conf', "restore_command = '{0}'".format(restore_command))
else:
node.append_conf(
'recovery.conf', "restore_command = '{0}'".format(restore_command))
node.slow_start()
result_new = node.safe_psql(
'postgres',
'select * from pgbench_accounts')
self.assertEqual(result, result_new)
with open(os.path.join(node.logs_dir, 'postgresql.log'), 'r') as f:
postgres_log_content = f.read()
# check that requesting of non-existing segment do not
# throwns aways prefetch
self.assertIn(
'pg_probackup archive-get failed to '
'deliver WAL file: 000000030000000000000006',
postgres_log_content)
self.assertIn(
'pg_probackup archive-get failed to '
'deliver WAL file: 000000020000000000000006',
postgres_log_content)
self.assertIn(
'pg_probackup archive-get used prefetched '
'WAL segment 000000010000000000000006, prefetch state: 5/10',
postgres_log_content)
# Clean after yourself
self.del_test_dir(module_name, fname)
def test_archive_get_batching_sanity(self):
"""
Make sure that batching works.
.gz file is corrupted and uncompressed is not, check that both
corruption detected and uncompressed file is used.
"""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
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()
self.backup_node(backup_dir, 'node', node, options=['--stream'])
node.pgbench_init(scale=50)
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(
backup_dir, 'node', replica, replica.data_dir)
self.set_replica(node, replica, log_shipping=True)
if node.major_version >= 12:
self.set_auto_conf(replica, {'restore_command': 'exit 1'})
else:
replica.append_conf('recovery.conf', "restore_command = 'exit 1'")
replica.slow_start(replica=True)
# at this point replica is consistent
restore_command = self.get_restore_command(backup_dir, 'node', replica)
restore_command += ' -j 2 --batch-size=10'
print(restore_command)
if node.major_version >= 12:
self.set_auto_conf(replica, {'restore_command': restore_command})
else:
replica.append_conf(
'recovery.conf', "restore_command = '{0}'".format(restore_command))
replica.restart()
sleep(5)
with open(os.path.join(replica.logs_dir, 'postgresql.log'), 'r') as f:
postgres_log_content = f.read()
self.assertIn(
'pg_probackup archive-get completed successfully, fetched: 10/10',
postgres_log_content)
self.assertIn('used prefetched WAL segment', postgres_log_content)
self.assertIn('prefetch state: 9/10', postgres_log_content)
self.assertIn('prefetch state: 8/10', postgres_log_content)
# Clean after yourself
self.del_test_dir(module_name, fname)
def test_archive_get_prefetch_corruption(self):
"""
Make sure that WAL corruption is detected.
And --prefetch-dir is honored.
"""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
initdb_params=['--data-checksums'],
pg_options={'autovacuum': 'off', 'wal_keep_segments': '200'})
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
self.backup_node(backup_dir, 'node', node, options=['--stream'])
node.pgbench_init(scale=50)
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(
backup_dir, 'node', replica, replica.data_dir)
self.set_replica(node, replica, log_shipping=True)
if node.major_version >= 12:
self.set_auto_conf(replica, {'restore_command': 'exit 1'})
else:
replica.append_conf('recovery.conf', "restore_command = 'exit 1'")
replica.slow_start(replica=True)
# at this point replica is consistent
restore_command = self.get_restore_command(backup_dir, 'node', replica)
restore_command += ' -j 2 --batch-size=10 --log-level-console=VERBOSE'
#restore_command += ' --batch-size=2 --log-level-console=VERBOSE'
if node.major_version >= 12:
self.set_auto_conf(replica, {'restore_command': restore_command})
else:
replica.append_conf(
'recovery.conf', "restore_command = '{0}'".format(restore_command))
replica.restart()
sleep(5)
with open(os.path.join(replica.logs_dir, 'postgresql.log'), 'r') as f:
postgres_log_content = f.read()
self.assertIn(
'pg_probackup archive-get completed successfully, fetched: 10/10',
postgres_log_content)
self.assertIn('used prefetched WAL segment', postgres_log_content)
self.assertIn('prefetch state: 9/10', postgres_log_content)
self.assertIn('prefetch state: 8/10', postgres_log_content)
replica.stop()
# generate WAL, copy it into prefetch directory, then corrupt
# some segment
node.pgbench_init(scale=20)
sleep(10)
# now copy WAL files into prefetch directory and corrupt some of them
archive_dir = os.path.join(backup_dir, 'wal', 'node')
files = os.listdir(archive_dir)
files.sort()
for filename in [files[-4], files[-3], files[-2], files[-1]]:
src_file = os.path.join(archive_dir, filename)
if node.major_version >= 10:
wal_dir = 'pg_wal'
else:
wal_dir = 'pg_xlog'
if filename.endswith('.gz'):
dst_file = os.path.join(replica.data_dir, wal_dir, 'pbk_prefetch', filename[:-3])
with gzip.open(src_file, 'rb') as f_in, open(dst_file, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out)
else:
dst_file = os.path.join(replica.data_dir, wal_dir, 'pbk_prefetch', filename)
shutil.copyfile(src_file, dst_file)
print(dst_file)
# corrupt file
if files[-2].endswith('.gz'):
filename = files[-2][:-3]
else:
filename = files[-2]
prefetched_file = os.path.join(replica.data_dir, wal_dir, 'pbk_prefetch', filename)
with open(prefetched_file, "rb+", 0) as f:
f.seek(8192*2)
f.write(b"SURIKEN")
f.flush()
f.close
# enable restore_command
restore_command = self.get_restore_command(backup_dir, 'node', replica)
restore_command += ' --batch-size=2 --log-level-console=VERBOSE'
if node.major_version >= 12:
self.set_auto_conf(replica, {'restore_command': restore_command})
else:
replica.append_conf(
'recovery.conf', "restore_command = '{0}'".format(restore_command))
os.remove(os.path.join(replica.logs_dir, 'postgresql.log'))
replica.slow_start(replica=True)
sleep(10)
with open(os.path.join(replica.logs_dir, 'postgresql.log'), 'r') as f:
postgres_log_content = f.read()
self.assertIn(
'Prefetched WAL segment {0} is invalid, cannot use it'.format(filename),
postgres_log_content)
self.assertIn(
'LOG: restored log file "{0}" from archive'.format(filename),
postgres_log_content)
# Clean after yourself
self.del_test_dir(module_name, fname)
# TODO test with multiple not archived segments.
# TODO corrupted file in archive.
# important - switchpoint may be NullOffset LSN and not actually existing in archive to boot.
# so write WAL validation code accordingly
tests/helpers/ptrack_helpers.py
+17 -13
View File
@@ -339,7 +339,7 @@ class ProbackupTest(object):
options['wal_level'] = 'logical'
options['hot_standby'] = 'off'
options['log_line_prefix'] = '"%t [%p]: [%l-1] "'
options['log_line_prefix'] = '%t [%p]: [%l-1] '
options['log_statement'] = 'none'
options['log_duration'] = 'on'
options['log_min_duration_statement'] = 0
@@ -1259,7 +1259,8 @@ class ProbackupTest(object):
def set_replica(
self, master, replica,
replica_name='replica',
synchronous=False
synchronous=False,
log_shipping=False
):
self.set_auto_conf(
@@ -1279,19 +1280,22 @@ class ProbackupTest(object):
if os.stat(probackup_recovery_path).st_size > 0:
config = 'probackup_recovery.conf'
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)
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')
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 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(