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

Merge branch 'master' into issue_174

Browse Source
This commit is contained in:
Grigory Smolkin
2020-03-15 19:18:44 +03:00
parent 1257184e77 e124234b1d
commit c32c6112d6
18 changed files with 1439 additions and 760 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/pgprobackup.xml
+14 -1
View File
@@ -3609,8 +3609,8 @@ pg_probackup restore -B <replaceable>backup_dir</replaceable> --instance <replac
[-R | --restore-as-replica] [--no-validate] [--skip-block-validation]
[--force] [--no-sync]
[--restore-command=<replaceable>cmdline</replaceable>]
[--restore-command=<replaceable>cmdline</replaceable>]
[--primary-conninfo=<replaceable>primary_conninfo</replaceable>]
[-S | --primary-slot-name=<replaceable>slotname</replaceable>]
[<replaceable>recovery_target_options</replaceable>] [<replaceable>logging_options</replaceable>] [<replaceable>remote_options</replaceable>]
[<replaceable>partial_restore_options</replaceable>] [<replaceable>remote_wal_archive_options</replaceable>]
</programlisting>
@@ -3663,6 +3663,19 @@ pg_probackup restore -B <replaceable>backup_dir</replaceable> --instance <replac
</listitem>
</varlistentry>
<varlistentry>
<term><option>-S</option></term>
<term><option>--primary-slot-name=<replaceable>slot_name</replaceable></option></term>
<listitem>
<para>
Sets the
<ulink url="https://postgrespro.com/docs/postgresql/current/runtime-config-replication#GUC-PRIMARY-SLOT-NAME">primary_slot_name</ulink>
parameter to the specified value.
This option will be ignored unless the <option>-R</option> flag if specified.
</para>
</listitem>
</varlistentry>
<varlistentry>
<term><option>-T <replaceable>OLDDIR</replaceable>=<replaceable>NEWDIR</replaceable></option></term>
<term><option>--tablespace-mapping=<replaceable>OLDDIR</replaceable>=<replaceable>NEWDIR</replaceable></option></term>
src/data.c
+282 -309
View File
@@ -192,77 +192,67 @@ parse_page(Page page, XLogRecPtr *lsn)
return false;
}
/* Read one page from file directly accessing disk
* return value:
* 2 - if the page is found but zeroed
* 1 - if the page is found and valid
* 0 - if the page is not found, probably truncated
* -1 - if the page is found but read size is not multiple of BLKSIZE
* -2 - if the page is found but page header is "insane"
* -3 - if the page is found but page checksumm is wrong
* -4 - something went wrong, check errno
*
/* We know that header is invalid, store specific
* details in errormsg.
*/
static int
read_page_from_file(pgFile *file, BlockNumber blknum,
FILE *in, Page page, XLogRecPtr *page_lsn,
uint32 checksum_version)
void
get_header_errormsg(Page page, char **errormsg)
{
off_t offset = blknum * BLCKSZ;
ssize_t read_len = 0;
PageHeader phdr = (PageHeader) page;
*errormsg = pgut_malloc(MAXPGPATH);
/* read the block */
read_len = fio_pread(in, page, offset);
if (PageGetPageSize(phdr) != BLCKSZ)
snprintf(*errormsg, MAXPGPATH, "page header invalid, "
"page size %lu is not equal to block size %u",
PageGetPageSize(phdr), BLCKSZ);
if (read_len != BLCKSZ)
{
else if (phdr->pd_lower < SizeOfPageHeaderData)
snprintf(*errormsg, MAXPGPATH, "page header invalid, "
"pd_lower %i is less than page header size %lu",
phdr->pd_lower, SizeOfPageHeaderData);
/* The block could have been truncated. It is fine. */
if (read_len == 0)
return 0;
else if (read_len > 0)
return -1;
else
return -4;
}
else if (phdr->pd_lower > phdr->pd_upper)
snprintf(*errormsg, MAXPGPATH, "page header invalid, "
"pd_lower %u is greater than pd_upper %u",
phdr->pd_lower, phdr->pd_upper);
/*
* If we found page with invalid header, at first check if it is zeroed,
* which is a valid state for page. If it is not, read it and check header
* again, because it's possible that we've read a partly flushed page.
* If after several attempts page header is still invalid, throw an error.
* The same idea is applied to checksum verification.
*/
if (!parse_page(page, page_lsn))
{
int i;
/* Check if the page is zeroed. */
for (i = 0; i < BLCKSZ && page[i] == 0; i++);
else if (phdr->pd_upper > phdr->pd_special)
snprintf(*errormsg, MAXPGPATH, "page header invalid, "
"pd_upper %u is greater than pd_special %u",
phdr->pd_upper, phdr->pd_special);
/* Page is zeroed. No need to check header and checksum. */
if (i == BLCKSZ)
return 2;
else if (phdr->pd_special > BLCKSZ)
snprintf(*errormsg, MAXPGPATH, "page header invalid, "
"pd_special %u is greater than block size %u",
phdr->pd_special, BLCKSZ);
return -2;
}
else if (phdr->pd_special != MAXALIGN(phdr->pd_special))
snprintf(*errormsg, MAXPGPATH, "page header invalid, "
"pd_special %i is misaligned, expected %lu",
phdr->pd_special, MAXALIGN(phdr->pd_special));
else if (phdr->pd_flags & ~PD_VALID_FLAG_BITS)
snprintf(*errormsg, MAXPGPATH, "page header invalid, "
"pd_flags mask contain illegal bits");
/* Verify checksum */
if (checksum_version)
{
BlockNumber blkno = file->segno * RELSEG_SIZE + blknum;
/*
* If checksum is wrong, sleep a bit and then try again
* several times. If it didn't help, throw error
*/
if (pg_checksum_page(page, blkno) != ((PageHeader) page)->pd_checksum)
return -3;
else
/* page header and checksum are correct */
return 1;
}
else
/* page header is correct and checksum check is disabled */
return 1;
snprintf(*errormsg, MAXPGPATH, "page header invalid");
}
/* We know that checksumms are mismatched, store specific
* details in errormsg.
*/
void
get_checksum_errormsg(Page page, char **errormsg, BlockNumber absolute_blkno)
{
PageHeader phdr = (PageHeader) page;
*errormsg = pgut_malloc(MAXPGPATH);
snprintf(*errormsg, MAXPGPATH,
"page verification failed, "
"calculated checksum %u but expected %u",
phdr->pd_checksum,
pg_checksum_page(page, absolute_blkno));
}
/*
@@ -273,15 +263,20 @@ read_page_from_file(pgFile *file, BlockNumber blknum,
* Prints appropriate warnings/errors/etc into log.
* Returns:
* PageIsOk(0) if page was successfully retrieved
* PageIsTruncated(-2) if the page was truncated
* SkipCurrentPage(-3) if we need to skip this page
* PageIsCorrupted(-4) if the page check mismatch
* PageIsTruncated(-1) if the page was truncated
* SkipCurrentPage(-2) if we need to skip this page,
* only used for DELTA backup
* PageIsCorrupted(-3) if the page checksum mismatch
* or header corruption,
* only used for checkdb
* TODO: probably we should always
* return it to the caller
*/
static int32
prepare_page(ConnectionArgs *conn_arg,
pgFile *file, XLogRecPtr prev_backup_start_lsn,
BlockNumber blknum, BlockNumber nblocks,
FILE *in, BackupMode backup_mode,
BlockNumber blknum, FILE *in,
BackupMode backup_mode,
Page page, bool strict,
uint32 checksum_version,
int ptrack_version_num,
@@ -289,9 +284,8 @@ prepare_page(ConnectionArgs *conn_arg,
const char *from_fullpath)
{
XLogRecPtr page_lsn = 0;
int try_again = 100;
int try_again = PAGE_READ_ATTEMPTS;
bool page_is_valid = false;
bool page_is_truncated = false;
BlockNumber absolute_blknum = file->segno * RELSEG_SIZE + blknum;
/* check for interrupt */
@@ -305,83 +299,104 @@ prepare_page(ConnectionArgs *conn_arg,
*/
if (backup_mode != BACKUP_MODE_DIFF_PTRACK || ptrack_version_num >= 20)
{
while (!page_is_valid && try_again)
int rc = 0;
while (!page_is_valid && try_again--)
{
int result = read_page_from_file(file, blknum, in, page,
&page_lsn, checksum_version);
/* read the block */
int read_len = fio_pread(in, page, blknum * BLCKSZ);
page_lsn = 0;
switch (result)
/* The block could have been truncated. It is fine. */
if (read_len == 0)
{
case 2:
elog(VERBOSE, "File: \"%s\" blknum %u, empty page", from_fullpath, blknum);
return PageIsOk;
elog(VERBOSE, "Cannot read block %u of \"%s\": "
"block truncated", blknum, from_fullpath);
return PageIsTruncated;
}
else if (read_len < 0)
elog(ERROR, "Cannot read block %u of \"%s\": %s",
blknum, from_fullpath, strerror(errno));
else if (read_len != BLCKSZ)
elog(WARNING, "Cannot read block %u of \"%s\": "
"read %i of %d, try again",
blknum, from_fullpath, read_len, BLCKSZ);
else
{
/* We have BLCKSZ of raw data, validate it */
rc = validate_one_page(page, absolute_blknum,
InvalidXLogRecPtr, &page_lsn,
checksum_version);
switch (rc)
{
case PAGE_IS_ZEROED:
elog(VERBOSE, "File: \"%s\" blknum %u, empty page", from_fullpath, blknum);
return PageIsOk;
case 1:
page_is_valid = true;
break;
case PAGE_IS_VALID:
/* in DELTA mode we must compare lsn */
if (backup_mode == BACKUP_MODE_DIFF_DELTA)
page_is_valid = true;
else
return PageIsOk;
break;
case 0:
/* This block was truncated.*/
page_is_truncated = true;
/* Page is not actually valid, but it is absent
* and we're not going to reread it or validate */
page_is_valid = true;
case PAGE_HEADER_IS_INVALID:
elog(VERBOSE, "File: \"%s\" blknum %u have wrong page header, try again",
from_fullpath, blknum);
break;
elog(VERBOSE, "File \"%s\", block %u, file was truncated",
from_fullpath, blknum);
break;
case -1:
elog(WARNING, "File: \"%s\", block %u, partial read, try again",
from_fullpath, blknum);
break;
case -2:
elog(LOG, "File: \"%s\" blknum %u have wrong page header, try again",
from_fullpath, blknum);
break;
case -3:
elog(LOG, "File: \"%s\" blknum %u have wrong checksum, try again",
from_fullpath, blknum);
break;
case -4:
elog(LOG, "File: \"%s\" access error: %s",
from_fullpath, strerror(errno));
break;
case PAGE_CHECKSUM_MISMATCH:
elog(VERBOSE, "File: \"%s\" blknum %u have wrong checksum, try again",
from_fullpath, blknum);
break;
default:
Assert(false);
}
}
/*
* If ptrack support is available use it to get invalid block
* If ptrack support is available, use it to get invalid block
* instead of rereading it 99 times
*/
if (result < 0 && strict && ptrack_version_num > 0)
if (!page_is_valid && strict && ptrack_version_num > 0)
{
elog(WARNING, "File \"%s\", block %u, try to fetch via shared buffer",
from_fullpath, blknum);
break;
}
try_again--;
}
/*
* If page is not valid after 100 attempts to read it
* throw an error.
*/
if (!page_is_valid &&
((strict && ptrack_version_num == 0) || !strict))
if (!page_is_valid)
{
/* show this message for checkdb, merge or backup without ptrack support */
elog(WARNING, "Corruption detected in file \"%s\", block %u",
from_fullpath, blknum);
}
int elevel = ERROR;
char *errormsg = NULL;
/* Backup with invalid block and without ptrack support must throw error */
if (!page_is_valid && strict && ptrack_version_num == 0)
elog(ERROR, "Data file corruption, canceling backup");
/* Get the details of corruption */
if (rc == PAGE_HEADER_IS_INVALID)
get_header_errormsg(page, &errormsg);
else if (rc == PAGE_CHECKSUM_MISMATCH)
get_checksum_errormsg(page, &errormsg,
file->segno * RELSEG_SIZE + blknum);
/* Error out in case of merge or backup without ptrack support;
* issue warning in case of checkdb or backup with ptrack support
*/
if (!strict || (strict && ptrack_version_num > 0))
elevel = WARNING;
if (errormsg)
elog(elevel, "Corruption detected in file \"%s\", block %u: %s",
from_fullpath, blknum, errormsg);
else
elog(elevel, "Corruption detected in file \"%s\", block %u",
from_fullpath, blknum);
pg_free(errormsg);
}
/* Checkdb not going futher */
if (!strict)
@@ -412,7 +427,7 @@ prepare_page(ConnectionArgs *conn_arg,
if (ptrack_page == NULL)
{
/* This block was truncated.*/
page_is_truncated = true;
return PageIsTruncated;
}
else if (page_size != BLCKSZ)
{
@@ -433,18 +448,15 @@ prepare_page(ConnectionArgs *conn_arg,
if (checksum_version)
((PageHeader) page)->pd_checksum = pg_checksum_page(page, absolute_blknum);
}
/* get lsn from page, provided by pg_ptrack_get_block() */
if (backup_mode == BACKUP_MODE_DIFF_DELTA &&
file->exists_in_prev &&
!page_is_truncated &&
!parse_page(page, &page_lsn))
elog(ERROR, "Cannot parse page after pg_ptrack_get_block. "
"Possible risk of a memory corruption");
}
if (page_is_truncated)
return PageIsTruncated;
/*
* Skip page if page lsn is less than START_LSN of parent backup.
* Nullified pages must be copied by DELTA backup, just to be safe.
@@ -475,10 +487,8 @@ compress_and_backup_page(pgFile *file, BlockNumber blknum,
const char *errormsg = NULL;
header.block = blknum;
header.compressed_size = page_state;
/* The page was not truncated, so we need to compress it */
/* Compress the page */
header.compressed_size = do_compress(compressed_page, sizeof(compressed_page),
page, BLCKSZ, calg, clevel,
&errormsg);
@@ -487,7 +497,7 @@ compress_and_backup_page(pgFile *file, BlockNumber blknum,
elog(WARNING, "An error occured during compressing block %u of file \"%s\": %s",
blknum, from_fullpath, errormsg);
file->compress_alg = calg;
file->compress_alg = calg; /* TODO: wtf? why here? */
/* The page was successfully compressed. */
if (header.compressed_size > 0 && header.compressed_size < BLCKSZ)
@@ -533,15 +543,15 @@ backup_data_file(ConnectionArgs* conn_arg, pgFile *file,
CompressAlg calg, int clevel, uint32 checksum_version,
int ptrack_version_num, const char *ptrack_schema, bool missing_ok)
{
FILE *in;
FILE *out;
BlockNumber blknum = 0;
BlockNumber nblocks = 0; /* number of blocks in file */
BlockNumber n_blocks_skipped = 0;
BlockNumber n_blocks_read = 0; /* number of blocks actually readed
* TODO: we should report them */
int page_state;
char curr_page[BLCKSZ];
FILE *in;
FILE *out;
BlockNumber blknum = 0;
BlockNumber nblocks = 0; /* number of blocks in source file */
BlockNumber n_blocks_skipped = 0;
int page_state;
char curr_page[BLCKSZ];
bool use_pagemap;
datapagemap_iterator_t *iter = NULL;
/* stdio buffers */
char in_buffer[STDIO_BUFSIZE];
@@ -633,113 +643,114 @@ backup_data_file(ConnectionArgs* conn_arg, pgFile *file,
* If page map is empty or file is not present in previous backup
* backup all pages of the relation.
*
* Usually enter here if backup_mode is FULL or DELTA.
* Also in some cases even PAGE backup is going here,
* becase not all data files are logged into WAL,
* for example CREATE DATABASE.
* Such files should be fully copied.
* In PTRACK 1.x there was a problem
* of data files with missing _ptrack map.
* Such files should be fully copied.
*/
if (file->pagemap.bitmapsize == PageBitmapIsEmpty ||
file->pagemap_isabsent || !file->exists_in_prev)
if (file->pagemap.bitmapsize == PageBitmapIsEmpty ||
file->pagemap_isabsent || !file->exists_in_prev ||
!file->pagemap.bitmap)
use_pagemap = false;
else
use_pagemap = true;
/* Remote mode */
if (fio_is_remote_file(in))
{
/* remote FULL and DELTA */
if (fio_is_remote_file(in))
char *errmsg = NULL;
BlockNumber err_blknum = 0;
/* TODO: retrying via ptrack should be implemented on the agent */
int rc = fio_send_pages(in, out, file,
/* send prev backup START_LSN */
backup_mode == BACKUP_MODE_DIFF_DELTA &&
file->exists_in_prev ? prev_backup_start_lsn : InvalidXLogRecPtr,
calg, clevel, checksum_version,
/* send pagemap if any */
use_pagemap ? &file->pagemap : NULL,
/* variables for error reporting */
&err_blknum, &errmsg);
/* check for errors */
if (rc == REMOTE_ERROR)
elog(ERROR, "Cannot read block %u of \"%s\": %s",
err_blknum, from_fullpath, strerror(errno));
else if (rc == PAGE_CORRUPTION)
{
int rc = fio_send_pages(in, out, file,
backup_mode == BACKUP_MODE_DIFF_DELTA &&
file->exists_in_prev ? prev_backup_start_lsn : InvalidXLogRecPtr,
&n_blocks_skipped, calg, clevel);
if (rc == PAGE_CHECKSUM_MISMATCH && ptrack_version_num >= 15)
/* only ptrack versions 1.5, 1.6, 1.7 and 2.x support this functionality */
goto RetryUsingPtrack;
if (rc < 0)
elog(ERROR, "Failed to read file \"%s\": %s",
from_fullpath,
rc == PAGE_CHECKSUM_MISMATCH ? "data file checksum mismatch" : strerror(-rc));
/* TODO: check that fio_send_pages ain`t lying about number of readed blocks */
n_blocks_read = rc;
file->read_size = n_blocks_read * BLCKSZ;
file->uncompressed_size = (n_blocks_read - n_blocks_skipped)*BLCKSZ;
if (errmsg)
elog(ERROR, "Corruption detected in file \"%s\", block %u: %s",
from_fullpath, err_blknum, errmsg);
else
elog(ERROR, "Corruption detected in file \"%s\", block %u",
from_fullpath, err_blknum);
}
else
{
/* local FULL and DELTA */
RetryUsingPtrack:
for (blknum = 0; blknum < nblocks; blknum++)
{
page_state = prepare_page(conn_arg, file, prev_backup_start_lsn,
blknum, nblocks, in, backup_mode,
curr_page, true, checksum_version,
ptrack_version_num, ptrack_schema,
from_fullpath);
if (page_state == PageIsTruncated)
break;
else if (rc == WRITE_FAILED)
elog(ERROR, "Cannot write block %u of \"%s\": %s",
err_blknum, to_fullpath, strerror(errno));
else if (page_state == SkipCurrentPage)
n_blocks_skipped++;
file->read_size = rc * BLCKSZ;
pg_free(errmsg);
else if (page_state == PageIsOk)
compress_and_backup_page(file, blknum, in, out, &(file->crc),
page_state, curr_page, calg, clevel,
from_fullpath, to_fullpath);
else
elog(ERROR, "Invalid page state: %i, file: %s, blknum %i",
page_state, file->rel_path, blknum);
n_blocks_read++;
file->read_size += BLCKSZ;
}
}
file->n_blocks = n_blocks_read;
}
/*
* If page map is not empty we scan only changed blocks.
*
* We will enter here if backup_mode is PAGE or PTRACK.
*/
/* Local mode */
else
{
datapagemap_iterator_t *iter;
iter = datapagemap_iterate(&file->pagemap);
while (datapagemap_next(iter, &blknum))
if (use_pagemap)
{
iter = datapagemap_iterate(&file->pagemap);
datapagemap_next(iter, &blknum); /* set first block */
}
while (blknum < nblocks)
{
page_state = prepare_page(conn_arg, file, prev_backup_start_lsn,
blknum, nblocks, in, backup_mode,
curr_page, true, checksum_version,
ptrack_version_num, ptrack_schema,
from_fullpath);
blknum, in, backup_mode, curr_page,
true, checksum_version,
ptrack_version_num, ptrack_schema,
from_fullpath);
if (page_state == PageIsTruncated)
break;
/* TODO: PAGE and PTRACK should never get SkipCurrentPage */
/* TODO: remove */
else if (page_state == SkipCurrentPage)
n_blocks_skipped++;
else if (page_state == PageIsOk)
compress_and_backup_page(file, blknum, in, out, &(file->crc),
page_state, curr_page, calg, clevel,
from_fullpath, to_fullpath);
page_state, curr_page, calg, clevel,
from_fullpath, to_fullpath);
/* TODO: handle PageIsCorrupted, currently it is done in prepare_page */
else
elog(ERROR, "Invalid page state: %i, file: %s, blknum %i",
page_state, file->rel_path, blknum);
Assert(false);
n_blocks_read++;
file->read_size += BLCKSZ;
}
pg_free(file->pagemap.bitmap);
pg_free(iter);
/* next block */
if (use_pagemap)
{
/* exit if pagemap is exhausted */
if (!datapagemap_next(iter, &blknum))
break;
}
else
blknum++;
}
}
pg_free(file->pagemap.bitmap);
pg_free(iter);
/* refresh n_blocks for FULL and DELTA */
if (backup_mode == BACKUP_MODE_FULL ||
backup_mode == BACKUP_MODE_DIFF_DELTA)
file->n_blocks = file->read_size / BLCKSZ;
if (fclose(out))
elog(ERROR, "Cannot close the backup file \"%s\": %s",
to_fullpath, strerror(errno));
@@ -1290,106 +1301,55 @@ create_empty_file(fio_location from_location, const char *to_root,
/*
* Validate given page.
*
* Returns value:
* 0 - if the page is not found
* 1 - if the page is found and valid
* -1 - if the page is found but invalid
* This function is expected to be executed multiple times,
* so avoid using elog within it.
* lsn from page is assigned to page_lsn pointer.
* TODO: switch to enum for return codes.
*/
#define PAGE_IS_NOT_FOUND 0
#define PAGE_IS_FOUND_AND_VALID 1
#define PAGE_IS_FOUND_AND_NOT_VALID -1
static int
validate_one_page(Page page, pgFile *file,
BlockNumber blknum, XLogRecPtr stop_lsn,
uint32 checksum_version)
int
validate_one_page(Page page, BlockNumber absolute_blkno,
XLogRecPtr stop_lsn, XLogRecPtr *page_lsn,
uint32 checksum_version)
{
PageHeader phdr;
XLogRecPtr lsn;
/* new level of paranoia */
if (page == NULL)
{
elog(LOG, "File \"%s\", block %u, page is NULL", file->path, blknum);
return PAGE_IS_NOT_FOUND;
}
phdr = (PageHeader) page;
if (PageIsNew(page))
/* check that page header is ok */
if (!parse_page(page, page_lsn))
{
int i;
int i;
/* Check if the page is zeroed. */
for(i = 0; i < BLCKSZ && page[i] == 0; i++);
for (i = 0; i < BLCKSZ && page[i] == 0; i++);
/* Page is zeroed. No need to verify checksums */
if (i == BLCKSZ)
{
elog(LOG, "File: %s blknum %u, page is New, empty zeroed page",
file->path, blknum);
return PAGE_IS_FOUND_AND_VALID;
}
else
{
elog(WARNING, "File: %s blknum %u, page is New, but not zeroed",
file->path, blknum);
}
return PAGE_IS_ZEROED;
/* Page is zeroed. No sense in checking header and checksum. */
return PAGE_IS_FOUND_AND_VALID;
/* Page does not looking good */
return PAGE_HEADER_IS_INVALID;
}
/* Verify checksum */
if (checksum_version)
{
/* Checksums are enabled, so check them. */
if (!(pg_checksum_page(page, file->segno * RELSEG_SIZE + blknum)
== ((PageHeader) page)->pd_checksum))
{
elog(WARNING, "File: %s blknum %u have wrong checksum",
file->path, blknum);
return PAGE_IS_FOUND_AND_NOT_VALID;
}
}
/* Check page for the sights of insanity.
* TODO: We should give more information about what exactly is looking "wrong"
*/
if (!(PageGetPageSize(phdr) == BLCKSZ &&
// PageGetPageLayoutVersion(phdr) == PG_PAGE_LAYOUT_VERSION &&
(phdr->pd_flags & ~PD_VALID_FLAG_BITS) == 0 &&
phdr->pd_lower >= SizeOfPageHeaderData &&
phdr->pd_lower <= phdr->pd_upper &&
phdr->pd_upper <= phdr->pd_special &&
phdr->pd_special <= BLCKSZ &&
phdr->pd_special == MAXALIGN(phdr->pd_special)))
{
/* Page does not looking good */
elog(WARNING, "Page header is looking insane: %s, block %i",
file->path, blknum);
return PAGE_IS_FOUND_AND_NOT_VALID;
if (pg_checksum_page(page, absolute_blkno) != ((PageHeader) page)->pd_checksum)
return PAGE_CHECKSUM_MISMATCH;
}
/* At this point page header is sane, if checksums are enabled - the`re ok.
* Check that page is not from future.
* Note, this check should be used only by validate command.
*/
if (stop_lsn > 0)
{
/* Get lsn from page header. Ensure that page is from our time. */
lsn = PageXLogRecPtrGet(phdr->pd_lsn);
if (lsn > stop_lsn)
{
elog(WARNING, "File: %s, block %u, checksum is %s. "
"Page is from future: pageLSN %X/%X stopLSN %X/%X",
file->path, blknum, checksum_version ? "correct" : "not enabled",
(uint32) (lsn >> 32), (uint32) lsn,
(uint32) (stop_lsn >> 32), (uint32) stop_lsn);
return PAGE_IS_FOUND_AND_NOT_VALID;
}
if (*page_lsn > stop_lsn)
return PAGE_LSN_FROM_FUTURE;
}
return PAGE_IS_FOUND_AND_VALID;
return PAGE_IS_VALID;
}
/*
@@ -1441,7 +1401,7 @@ check_data_file(ConnectionArgs *arguments, pgFile *file,
{
page_state = prepare_page(NULL, file, InvalidXLogRecPtr,
blknum, nblocks, in, BACKUP_MODE_FULL,
blknum, in, BACKUP_MODE_FULL,
curr_page, false, checksum_version,
0, NULL, from_fullpath);
@@ -1456,19 +1416,6 @@ check_data_file(ConnectionArgs *arguments, pgFile *file,
is_valid = false;
continue;
}
/* At this point page is found and its checksum is ok, if any
* but could be 'insane'
* TODO: between prepare_page and validate_one_page we
* compute and compare checksum twice, it`s ineffective
*/
if (validate_one_page(curr_page, file, blknum,
InvalidXLogRecPtr,
0) == PAGE_IS_FOUND_AND_NOT_VALID)
{
/* Page is corrupted */
is_valid = false;
}
}
fclose(in);
@@ -1507,10 +1454,12 @@ check_file_pages(pgFile *file, XLogRecPtr stop_lsn, uint32 checksum_version,
/* read and validate pages one by one */
while (true)
{
int rc = 0;
DataPage compressed_page; /* used as read buffer */
DataPage page;
BackupPageHeader header;
BlockNumber blknum = 0;
XLogRecPtr page_lsn = 0;
if (interrupted || thread_interrupted)
elog(ERROR, "Interrupted during data file validation");
@@ -1597,15 +1546,39 @@ check_file_pages(pgFile *file, XLogRecPtr stop_lsn, uint32 checksum_version,
return false;
}
if (validate_one_page(page.data, file, blknum,
stop_lsn, checksum_version) == PAGE_IS_FOUND_AND_NOT_VALID)
is_valid = false;
rc = validate_one_page(page.data,
file->segno * RELSEG_SIZE + blknum,
stop_lsn, &page_lsn, checksum_version);
}
else
rc = validate_one_page(compressed_page.data,
file->segno * RELSEG_SIZE + blknum,
stop_lsn, &page_lsn, checksum_version);
switch (rc)
{
if (validate_one_page(compressed_page.data, file, blknum,
stop_lsn, checksum_version) == PAGE_IS_FOUND_AND_NOT_VALID)
case PAGE_IS_NOT_FOUND:
elog(LOG, "File \"%s\", block %u, page is NULL", file->rel_path, blknum);
break;
case PAGE_IS_ZEROED:
elog(LOG, "File: %s blknum %u, empty zeroed page", file->rel_path, blknum);
break;
case PAGE_HEADER_IS_INVALID:
elog(WARNING, "Page header is looking insane: %s, block %i", file->rel_path, blknum);
is_valid = false;
break;
case PAGE_CHECKSUM_MISMATCH:
elog(WARNING, "File: %s blknum %u have wrong checksum", file->rel_path, blknum);
is_valid = false;
break;
case PAGE_LSN_FROM_FUTURE:
elog(WARNING, "File: %s, block %u, checksum is %s. "
"Page is from future: pageLSN %X/%X stopLSN %X/%X",
file->rel_path, blknum,
checksum_version ? "correct" : "not enabled",
(uint32) (page_lsn >> 32), (uint32) page_lsn,
(uint32) (stop_lsn >> 32), (uint32) stop_lsn);
break;
}
}
src/dir.c
+1 -2
View File
@@ -1759,7 +1759,6 @@ write_database_map(pgBackup *backup, parray *database_map, parray *backup_files_
char database_dir[MAXPGPATH];
char database_map_path[MAXPGPATH];
// pgBackupGetPath(backup, path, lengthof(path), DATABASE_DIR);
join_path_components(database_dir, backup->root_dir, DATABASE_DIR);
join_path_components(database_map_path, database_dir, DATABASE_MAP);
@@ -1783,7 +1782,7 @@ write_database_map(pgBackup *backup, parray *database_map, parray *backup_files_
file->path = pgut_strdup(DATABASE_MAP);
file->crc = pgFileGetCRC(database_map_path, true, false);
file->write_size = file->read_size;
file->write_size = file->size;
file->uncompressed_size = file->read_size;
parray_append(backup_files_list, file);
}
src/help.c
+31 -22
View File
@@ -153,6 +153,8 @@ help_pg_probackup(void)
printf(_(" [--recovery-target-name=target-name]\n"));
printf(_(" [--recovery-target-action=pause|promote|shutdown]\n"));
printf(_(" [--restore-as-replica] [--force]\n"));
printf(_(" [--primary-conninfo=primary_conninfo]\n"));
printf(_(" [-S | --primary-slot-name=slotname]\n"));
printf(_(" [--no-validate] [--skip-block-validation]\n"));
printf(_(" [-T OLDDIR=NEWDIR] [--progress]\n"));
printf(_(" [--external-mapping=OLDDIR=NEWDIR]\n"));
@@ -383,20 +385,20 @@ help_restore(void)
{
printf(_("\n%s restore -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" [-D pgdata-path] [-i backup-id] [-j num-threads]\n"));
printf(_(" [--progress] [--force] [--no-sync]\n"));
printf(_(" [--no-validate] [--skip-block-validation]\n"));
printf(_(" [-T OLDDIR=NEWDIR]\n"));
printf(_(" [--external-mapping=OLDDIR=NEWDIR]\n"));
printf(_(" [--skip-external-dirs]\n"));
printf(_(" [--db-include dbname | --db-exclude dbname]\n"));
printf(_(" [--recovery-target-time=time|--recovery-target-xid=xid\n"));
printf(_(" |--recovery-target-lsn=lsn [--recovery-target-inclusive=boolean]]\n"));
printf(_(" [--recovery-target-timeline=timeline]\n"));
printf(_(" [--recovery-target=immediate|latest]\n"));
printf(_(" [--recovery-target-name=target-name]\n"));
printf(_(" [--recovery-target-action=pause|promote|shutdown]\n"));
printf(_(" [--restore-as-replica] [--force]\n"));
printf(_(" [--no-validate] [--skip-block-validation]\n"));
printf(_(" [-T OLDDIR=NEWDIR] [--progress]\n"));
printf(_(" [--external-mapping=OLDDIR=NEWDIR]\n"));
printf(_(" [--skip-external-dirs]\n"));
printf(_(" [--restore-command=cmdline]\n"));
printf(_(" [--no-sync]\n"));
printf(_(" [--db-include dbname | --db-exclude dbname]\n"));
printf(_(" [-R | --restore-as-replica]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
@@ -411,6 +413,22 @@ help_restore(void)
printf(_(" -j, --threads=NUM number of parallel threads\n"));
printf(_(" --progress show progress\n"));
printf(_(" --force ignore invalid status of the restored backup\n"));
printf(_(" --no-sync do not sync restored files to disk\n"));
printf(_(" --no-validate disable backup validation during restore\n"));
printf(_(" --skip-block-validation set to validate only file-level checksum\n"));
printf(_(" -T, --tablespace-mapping=OLDDIR=NEWDIR\n"));
printf(_(" relocate the tablespace from directory OLDDIR to NEWDIR\n"));
printf(_(" --external-mapping=OLDDIR=NEWDIR\n"));
printf(_(" relocate the external directory from OLDDIR to NEWDIR\n"));
printf(_(" --skip-external-dirs do not restore all external directories\n"));
printf(_("\n Partial restore options:\n"));
printf(_(" --db-include dbname restore only specified databases\n"));
printf(_(" --db-exclude dbname do not restore specified databases\n"));
printf(_("\n Recovery options:\n"));
printf(_(" --recovery-target-time=time time stamp up to which recovery will proceed\n"));
printf(_(" --recovery-target-xid=xid transaction ID up to which recovery will proceed\n"));
printf(_(" --recovery-target-lsn=lsn LSN of the write-ahead log location up to which recovery will proceed\n"));
@@ -425,24 +443,15 @@ help_restore(void)
printf(_(" --recovery-target-action=pause|promote|shutdown\n"));
printf(_(" action the server should take once the recovery target is reached\n"));
printf(_(" (default: pause)\n"));
printf(_(" --restore-command=cmdline command to use as 'restore_command' in recovery.conf; 'none' disables\n"));
printf(_("\n Standby options:\n"));
printf(_(" -R, --restore-as-replica write a minimal recovery.conf in the output directory\n"));
printf(_(" to ease setting up a standby server\n"));
printf(_(" --force ignore invalid status of the restored backup\n"));
printf(_(" --no-validate disable backup validation during restore\n"));
printf(_(" --skip-block-validation set to validate only file-level checksum\n"));
printf(_(" -T, --tablespace-mapping=OLDDIR=NEWDIR\n"));
printf(_(" relocate the tablespace from directory OLDDIR to NEWDIR\n"));
printf(_(" --external-mapping=OLDDIR=NEWDIR\n"));
printf(_(" relocate the external directory from OLDDIR to NEWDIR\n"));
printf(_(" --skip-external-dirs do not restore all external directories\n"));
printf(_(" --restore-command=cmdline command to use as 'restore_command' in recovery.conf; 'none' disables\n"));
printf(_(" --no-sync do not sync restored files to disk\n"));
printf(_("\n Partial restore options:\n"));
printf(_(" --db-include dbname restore only specified databases\n"));
printf(_(" --db-exclude dbname do not restore specified databases\n"));
printf(_(" --primary-conninfo=primary_conninfo\n"));
printf(_(" connection string to be used for establishing connection\n"));
printf(_(" with the primary server\n"));
printf(_(" -S, --primary-slot-name=slotname replication slot to be used for WAL streaming from the primary server\n"));
printf(_("\n Logging options:\n"));
printf(_(" --log-level-console=log-level-console\n"));
src/init.c
+17
View File
@@ -103,6 +103,23 @@ do_add_instance(InstanceConfig *instance)
SOURCE_FILE);
config_set_opt(instance_options, &instance->xlog_seg_size,
SOURCE_FILE);
/* Kludge: do not save remote options into config */
config_set_opt(instance_options, &instance_config.remote.host,
SOURCE_DEFAULT);
config_set_opt(instance_options, &instance_config.remote.proto,
SOURCE_DEFAULT);
config_set_opt(instance_options, &instance_config.remote.port,
SOURCE_DEFAULT);
config_set_opt(instance_options, &instance_config.remote.path,
SOURCE_DEFAULT);
config_set_opt(instance_options, &instance_config.remote.user,
SOURCE_DEFAULT);
config_set_opt(instance_options, &instance_config.remote.ssh_options,
SOURCE_DEFAULT);
config_set_opt(instance_options, &instance_config.remote.ssh_config,
SOURCE_DEFAULT);
/* pgdata was set through command line */
do_set_config(true);
src/pg_probackup.c
+5
View File
@@ -174,6 +174,7 @@ static ConfigOption cmd_options[] =
{ 'f', 'b', "backup-mode", opt_backup_mode, SOURCE_CMD_STRICT },
{ 'b', 'C', "smooth-checkpoint", &smooth_checkpoint, SOURCE_CMD_STRICT },
{ 's', 'S', "slot", &replication_slot, SOURCE_CMD_STRICT },
{ 's', 'S', "primary-slot-name",&replication_slot, SOURCE_CMD_STRICT },
{ 'b', 181, "temp-slot", &temp_slot, SOURCE_CMD_STRICT },
{ 'b', 182, "delete-wal", &delete_wal, SOURCE_CMD_STRICT },
{ 'b', 183, "delete-expired", &delete_expired, SOURCE_CMD_STRICT },
@@ -683,12 +684,16 @@ main(int argc, char *argv[])
if (force)
no_validate = true;
if (replication_slot != NULL)
restore_as_replica = true;
/* keep all params in one structure */
restore_params = pgut_new(pgRestoreParams);
restore_params->is_restore = (backup_subcmd == RESTORE_CMD);
restore_params->force = force;
restore_params->no_validate = no_validate;
restore_params->restore_as_replica = restore_as_replica;
restore_params->primary_slot_name = replication_slot;
restore_params->skip_block_validation = skip_block_validation;
restore_params->skip_external_dirs = skip_external_dirs;
restore_params->partial_db_list = NULL;
src/pg_probackup.h
+36 -4
View File
@@ -85,7 +85,10 @@ extern const char *PROGRAM_EMAIL;
#endif
/* stdio buffer size */
#define STDIO_BUFSIZE 65536
#define STDIO_BUFSIZE 65536
/* retry attempts */
#define PAGE_READ_ATTEMPTS 100
/* Check if an XLogRecPtr value is pointed to 0 offset */
#define XRecOffIsNull(xlrp) \
@@ -169,7 +172,8 @@ typedef struct pgFile
bool exists_in_prev; /* Mark files, both data and regular, that exists in previous backup */
CompressAlg compress_alg; /* compression algorithm applied to the file */
volatile pg_atomic_flag lock;/* lock for synchronization of parallel threads */
datapagemap_t pagemap; /* bitmap of pages updated since previous backup */
datapagemap_t pagemap; /* bitmap of pages updated since previous backup
may take up to 16kB per file */
bool pagemap_isabsent; /* Used to mark files with unknown state of pagemap,
* i.e. datafiles without _ptrack */
} pgFile;
@@ -418,6 +422,7 @@ typedef struct pgRestoreParams
bool skip_external_dirs;
bool skip_block_validation; //Start using it
const char *restore_command;
const char *primary_slot_name;
/* options for partial restore */
PartialRestoreType partial_restore_type;
@@ -525,9 +530,9 @@ typedef struct BackupPageHeader
/* Special value for compressed_size field */
#define PageIsOk 0
#define SkipCurrentPage -1
#define PageIsTruncated -2
#define SkipCurrentPage -3
#define PageIsCorrupted -4 /* used by checkdb */
#define PageIsCorrupted -3 /* used by checkdb */
/*
@@ -728,6 +733,18 @@ extern void help_command(char *command);
/* in validate.c */
extern void pgBackupValidate(pgBackup* backup, pgRestoreParams *params);
extern int do_validate_all(void);
extern int validate_one_page(Page page, BlockNumber absolute_blkno,
XLogRecPtr stop_lsn, XLogRecPtr *page_lsn,
uint32 checksum_version);
/* return codes for validate_one_page */
/* TODO: use enum */
#define PAGE_IS_VALID (-1)
#define PAGE_IS_NOT_FOUND (-2)
#define PAGE_IS_ZEROED (-3)
#define PAGE_HEADER_IS_INVALID (-4)
#define PAGE_CHECKSUM_MISMATCH (-5)
#define PAGE_LSN_FROM_FUTURE (-6)
/* in catalog.c */
extern pgBackup *read_backup(const char *instance_name, time_t timestamp);
@@ -952,4 +969,19 @@ extern char *pg_ptrack_get_and_clear(Oid tablespace_oid,
extern XLogRecPtr get_last_ptrack_lsn(PGconn *backup_conn, PGNodeInfo *nodeInfo);
extern parray * pg_ptrack_get_pagemapset(PGconn *backup_conn, const char *ptrack_schema, XLogRecPtr lsn);
/* FIO */
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)
extern void get_header_errormsg(Page page, char **errormsg);
extern void get_checksum_errormsg(Page page, char **errormsg,
BlockNumber absolute_blkno);
#endif /* PG_PROBACKUP_H */
src/restore.c
+13 -5
View File
@@ -472,6 +472,9 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
create_recovery_conf(target_backup_id, rt, dest_backup, params);
}
/* ssh connection to longer needed */
fio_disconnect();
/* cleanup */
parray_walk(backups, pgBackupFree);
parray_free(backups);
@@ -977,6 +980,7 @@ create_recovery_conf(time_t backup_id,
/* construct restore_command */
if (pitr_requested)
{
fio_fprintf(fp, "\n## recovery settings\n");
/* If restore_command is provided, use it. Otherwise construct it from scratch. */
if (restore_command_provided)
sprintf(restore_command_guc, "%s", instance_config.restore_command);
@@ -1052,8 +1056,15 @@ create_recovery_conf(time_t backup_id,
fio_fprintf(fp, "recovery_target_action = '%s'\n", "pause");
}
if (pitr_requested)
{
elog(LOG, "Setting restore_command to '%s'", restore_command_guc);
fio_fprintf(fp, "restore_command = '%s'\n", restore_command_guc);
}
if (params->restore_as_replica)
{
fio_fprintf(fp, "\n## standby settings\n");
/* standby_mode was removed in PG12 */
#if PG_VERSION_NUM < 120000
fio_fprintf(fp, "standby_mode = 'on'\n");
@@ -1063,12 +1074,9 @@ create_recovery_conf(time_t backup_id,
fio_fprintf(fp, "primary_conninfo = '%s'\n", params->primary_conninfo);
else if (backup->primary_conninfo)
fio_fprintf(fp, "primary_conninfo = '%s'\n", backup->primary_conninfo);
}
if (pitr_requested)
{
elog(LOG, "Setting restore_command to '%s'", restore_command_guc);
fio_fprintf(fp, "restore_command = '%s'\n", restore_command_guc);
if (params->primary_slot_name != NULL)
fio_fprintf(fp, "primary_slot_name = '%s'\n", params->primary_slot_name);
}
if (fio_fflush(fp) != 0 ||
src/utils/file.c
+235 -88
View File
@@ -14,7 +14,6 @@
#define PRINTF_BUF_SIZE 1024
#define FILE_PERMISSIONS 0600
#define PAGE_READ_ATTEMPTS 100
static __thread unsigned long fio_fdset = 0;
static __thread void* fio_stdin_buffer;
@@ -27,11 +26,12 @@ fio_location MyLocation;
typedef struct
{
BlockNumber nblocks;
BlockNumber segBlockNum;
BlockNumber segmentno;
XLogRecPtr horizonLsn;
uint32 checksumVersion;
int calg;
int clevel;
int bitmapsize;
} fio_send_request;
@@ -114,6 +114,7 @@ fio_safestat(const char *path, struct stat *buf)
#define stat(x, y) fio_safestat(x, y)
/* TODO: use real pread on Linux */
static ssize_t pread(int fd, void* buf, size_t size, off_t off)
{
off_t rc = lseek(fd, off, SEEK_SET);
@@ -371,6 +372,12 @@ fio_disconnect(void)
{
if (fio_stdin)
{
fio_header hdr;
hdr.cop = FIO_DISCONNECT;
hdr.size = 0;
IO_CHECK(fio_write_all(fio_stdout, &hdr, sizeof(hdr)), sizeof(hdr));
IO_CHECK(fio_read_all(fio_stdin, &hdr, sizeof(hdr)), sizeof(hdr));
Assert(hdr.cop == FIO_DISCONNECTED);
SYS_CHECK(close(fio_stdin));
SYS_CHECK(close(fio_stdout));
fio_stdin = 0;
@@ -549,13 +556,14 @@ int fio_pread(FILE* f, void* buf, off_t offs)
if (hdr.size != 0)
IO_CHECK(fio_read_all(fio_stdin, buf, hdr.size), hdr.size);
/* TODO: error handling */
return hdr.arg;
}
else
{
/* For local file, opened by fopen, we should use stdio operations */
int rc;
rc = fseek(f, offs, SEEK_SET);
/* For local file, opened by fopen, we should use stdio functions */
int rc = fseek(f, offs, SEEK_SET);
if (rc < 0)
return rc;
@@ -1323,8 +1331,24 @@ static void fio_send_file(int out, char const* path)
}
}
int fio_send_pages(FILE* in, FILE* out, pgFile *file,
XLogRecPtr horizonLsn, BlockNumber* nBlocksSkipped, int calg, int clevel)
/*
* Return number of actually(!) readed blocks, attempts or
* half-readed block are not counted.
* Return values in case of error:
* REMOTE_ERROR
* PAGE_CORRUPTION
* WRITE_FAILED
*
* If none of the above, this function return number of blocks
* readed by remote agent.
*
* In case of DELTA mode horizonLsn must be a valid lsn,
* otherwise it should be set to InvalidXLogRecPtr.
*/
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)
{
struct {
fio_header hdr;
@@ -1335,144 +1359,240 @@ int fio_send_pages(FILE* in, FILE* out, pgFile *file,
Assert(fio_is_remote_file(in));
req.hdr.cop = FIO_SEND_PAGES;
req.hdr.size = sizeof(fio_send_request);
/* send message with header
8bytes 20bytes var
------------------------------------------------------
| fio_header | fio_send_request | BITMAP(if any) |
------------------------------------------------------
*/
req.hdr.handle = fio_fileno(in) & ~FIO_PIPE_MARKER;
if (pagemap)
{
req.hdr.cop = FIO_SEND_PAGES_PAGEMAP;
req.hdr.size = sizeof(fio_send_request) + pagemap->bitmapsize;
req.arg.bitmapsize = pagemap->bitmapsize;
/* TODO: add optimization for the case of pagemap
* containing small number of blocks with big serial numbers:
* https://github.com/postgrespro/pg_probackup/blob/remote_page_backup/src/utils/file.c#L1211
*/
}
else
{
req.hdr.cop = FIO_SEND_PAGES;
req.hdr.size = sizeof(fio_send_request);
}
req.arg.nblocks = file->size/BLCKSZ;
req.arg.segBlockNum = file->segno * RELSEG_SIZE;
req.arg.segmentno = file->segno * RELSEG_SIZE;
req.arg.horizonLsn = horizonLsn;
req.arg.checksumVersion = current.checksum_version;
req.arg.checksumVersion = checksum_version;
req.arg.calg = calg;
req.arg.clevel = clevel;
file->compress_alg = calg;
file->compress_alg = calg; /* TODO: wtf? why here? */
//<-----
// datapagemap_iterator_t *iter;
// BlockNumber blkno;
// iter = datapagemap_iterate(pagemap);
// while (datapagemap_next(iter, &blkno))
// elog(INFO, "block %u", blkno);
// pg_free(iter);
//<-----
IO_CHECK(fio_write_all(fio_stdout, &req, sizeof(req)), sizeof(req));
if (pagemap)
/* now send pagemap itself */
IO_CHECK(fio_write_all(fio_stdout, pagemap->bitmap, pagemap->bitmapsize), pagemap->bitmapsize);
while (true)
{
fio_header hdr;
char buf[BLCKSZ + sizeof(BackupPageHeader)];
IO_CHECK(fio_read_all(fio_stdin, &hdr, sizeof(hdr)), sizeof(hdr));
Assert(hdr.cop == FIO_PAGE);
if ((int)hdr.arg < 0) /* read error */
if (interrupted)
elog(ERROR, "Interrupted during page reading");
if (hdr.cop == FIO_ERROR)
{
return (int)hdr.arg;
errno = hdr.arg;
*err_blknum = hdr.size;
return REMOTE_ERROR;
}
blknum = hdr.arg;
if (hdr.size == 0) /* end of segment */
break;
Assert(hdr.size <= sizeof(buf));
IO_CHECK(fio_read_all(fio_stdin, buf, hdr.size), hdr.size);
COMP_FILE_CRC32(true, file->crc, buf, hdr.size);
if (fio_fwrite(out, buf, hdr.size) != hdr.size)
else if (hdr.cop == FIO_SEND_FILE_CORRUPTION)
{
int errno_tmp = errno;
fio_fclose(out);
elog(ERROR, "File: %s, cannot write backup at block %u: %s",
file->path, blknum, strerror(errno_tmp));
*err_blknum = hdr.arg;
if (hdr.size > 0)
{
IO_CHECK(fio_read_all(fio_stdin, buf, hdr.size), hdr.size);
*errormsg = pgut_malloc(hdr.size);
strncpy(*errormsg, buf, hdr.size);
}
return PAGE_CORRUPTION;
}
file->write_size += hdr.size;
n_blocks_read++;
if (((BackupPageHeader*)buf)->compressed_size == PageIsTruncated)
else if (hdr.cop == FIO_SEND_FILE_EOF)
{
blknum += 1;
/* n_blocks_read reported by EOF */
n_blocks_read = hdr.size;
break;
}
else if (hdr.cop == FIO_PAGE)
{
blknum = hdr.arg;
Assert(hdr.size <= sizeof(buf));
IO_CHECK(fio_read_all(fio_stdin, buf, hdr.size), hdr.size);
COMP_FILE_CRC32(true, file->crc, buf, hdr.size);
if (fio_fwrite(out, buf, hdr.size) != hdr.size)
{
fio_fclose(out);
*err_blknum = blknum;
return WRITE_FAILED;
}
file->write_size += hdr.size;
file->uncompressed_size += BLCKSZ;
}
else
elog(ERROR, "Remote agent returned message of unknown type");
}
*nBlocksSkipped = blknum - n_blocks_read;
return blknum;
return n_blocks_read;
}
static void fio_send_pages_impl(int fd, int out, fio_send_request* req)
static void fio_send_pages_impl(int fd, int out, char* buf, bool with_pagemap)
{
BlockNumber blknum;
BlockNumber blknum = 0;
BlockNumber n_blocks_read = 0;
XLogRecPtr page_lsn = 0;
char read_buffer[BLCKSZ+1];
fio_header hdr;
fio_send_request *req = (fio_send_request*) buf;
/* parse buffer */
datapagemap_t *map = NULL;
datapagemap_iterator_t *iter = NULL;
if (with_pagemap)
{
map = pgut_malloc(sizeof(datapagemap_t));
map->bitmapsize = req->bitmapsize;
map->bitmap = (char*) buf + sizeof(fio_send_request);
/* get first block */
iter = datapagemap_iterate(map);
datapagemap_next(iter, &blknum);
}
hdr.cop = FIO_PAGE;
read_buffer[BLCKSZ] = 1; /* barrier */
for (blknum = 0; blknum < req->nblocks; blknum++)
while (blknum < req->nblocks)
{
int rc = 0;
int retry_attempts = PAGE_READ_ATTEMPTS;
XLogRecPtr page_lsn = InvalidXLogRecPtr;
while (true)
/* TODO: handle signals on the agent */
if (interrupted)
elog(ERROR, "Interrupted during remote page reading");
/* read page, check header and validate checksumms */
/* TODO: libpq connection on the agent, so we can do ptrack
* magic right here.
*/
for (;;)
{
ssize_t rc = pread(fd, read_buffer, BLCKSZ, blknum*BLCKSZ);
ssize_t read_len = pread(fd, read_buffer, BLCKSZ, blknum*BLCKSZ);
page_lsn = InvalidXLogRecPtr;
if (rc <= 0)
/* report eof */
if (read_len == 0)
goto eof;
/* report error */
else if (read_len < 0)
{
if (rc < 0)
{
hdr.arg = -errno;
hdr.size = 0;
Assert((int)hdr.arg < 0);
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
}
else
{
BackupPageHeader bph;
bph.block = blknum;
bph.compressed_size = PageIsTruncated;
hdr.arg = blknum;
hdr.size = sizeof(bph);
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
IO_CHECK(fio_write_all(out, &bph, sizeof(bph)), sizeof(bph));
}
return;
/* TODO: better to report exact error message, not errno */
hdr.cop = FIO_ERROR;
hdr.arg = errno;
hdr.size = blknum;
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
goto cleanup;
}
else if (rc == BLCKSZ)
else if (read_len == BLCKSZ)
{
if (!parse_page((Page)read_buffer, &page_lsn))
{
int i;
for (i = 0; read_buffer[i] == 0; i++);
rc = validate_one_page(read_buffer, req->segmentno + blknum,
InvalidXLogRecPtr, &page_lsn, req->checksumVersion);
/* Page is zeroed. No need to check header and checksum. */
if (i == BLCKSZ)
break;
}
else if (!req->checksumVersion
|| pg_checksum_page(read_buffer, req->segBlockNum + blknum) == ((PageHeader)read_buffer)->pd_checksum)
{
/* TODO: optimize copy of zeroed page */
if (rc == PAGE_IS_ZEROED)
break;
else if (rc == PAGE_IS_VALID)
break;
}
}
// else /* readed less than BLKSZ bytes, retry */
/* File is either has insane header or invalid checksum,
* retry. If retry attempts are exhausted, report corruption.
*/
if (--retry_attempts == 0)
{
hdr.size = 0;
hdr.arg = PAGE_CHECKSUM_MISMATCH;
char *errormsg = NULL;
hdr.cop = FIO_SEND_FILE_CORRUPTION;
hdr.arg = blknum;
/* Construct the error message */
if (rc == PAGE_HEADER_IS_INVALID)
get_header_errormsg(read_buffer, &errormsg);
else if (rc == PAGE_CHECKSUM_MISMATCH)
get_checksum_errormsg(read_buffer, &errormsg,
req->segmentno + blknum);
/* if error message is not empty, set payload size to its length */
hdr.size = errormsg ? strlen(errormsg) + 1 : 0;
/* send header */
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
return;
/* send error message if any */
if (errormsg)
IO_CHECK(fio_write_all(out, errormsg, hdr.size), hdr.size);
pg_free(errormsg);
goto cleanup;
}
}
n_blocks_read++;
/*
* horizonLsn is not 0 for delta backup.
* horizonLsn is not 0 only in case of delta backup.
* As far as unsigned number are always greater or equal than zero,
* there is no sense to add more checks.
*/
if (page_lsn >= req->horizonLsn || page_lsn == InvalidXLogRecPtr)
if ((req->horizonLsn == InvalidXLogRecPtr) ||
(page_lsn == InvalidXLogRecPtr) || /* zeroed page */
(req->horizonLsn > 0 && page_lsn >= req->horizonLsn)) /* delta */
{
char write_buffer[BLCKSZ*2];
BackupPageHeader* bph = (BackupPageHeader*)write_buffer;
const char *errormsg = NULL;
/* compress page */
hdr.arg = bph->block = blknum;
hdr.size = sizeof(BackupPageHeader);
bph->compressed_size = do_compress(write_buffer + sizeof(BackupPageHeader), sizeof(write_buffer) - sizeof(BackupPageHeader),
bph->compressed_size = do_compress(write_buffer + sizeof(BackupPageHeader),
sizeof(write_buffer) - sizeof(BackupPageHeader),
read_buffer, BLCKSZ, req->calg, req->clevel,
&errormsg);
NULL);
if (bph->compressed_size <= 0 || bph->compressed_size >= BLCKSZ)
{
/* Do not compress page */
@@ -1484,10 +1604,29 @@ static void fio_send_pages_impl(int fd, int out, fio_send_request* req)
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
IO_CHECK(fio_write_all(out, write_buffer, hdr.size), hdr.size);
}
/* next block */
if (with_pagemap)
{
/* exit if pagemap is exhausted */
if (!datapagemap_next(iter, &blknum))
break;
}
else
blknum++;
}
hdr.size = 0;
hdr.arg = blknum;
eof:
/* We are done, send eof */
hdr.cop = FIO_SEND_FILE_EOF;
hdr.arg = 0;
hdr.size = n_blocks_read; /* TODO: report number of backed up blocks */
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
cleanup:
pg_free(map);
pg_free(iter);
return;
}
/* Execute commands at remote host */
@@ -1514,7 +1653,7 @@ void fio_communicate(int in, int out)
SYS_CHECK(setmode(out, _O_BINARY));
#endif
/* Main loop until command of processing master command */
/* Main loop until end of processing all master commands */
while ((rc = fio_read_all(in, &hdr, sizeof hdr)) == sizeof(hdr)) {
if (hdr.size != 0) {
if (hdr.size > buf_size) {
@@ -1631,10 +1770,14 @@ void fio_communicate(int in, int out)
break;
case FIO_SEND_PAGES:
Assert(hdr.size == sizeof(fio_send_request));
fio_send_pages_impl(fd[hdr.handle], out, (fio_send_request*)buf);
fio_send_pages_impl(fd[hdr.handle], out, buf, false);
break;
case FIO_SEND_PAGES_PAGEMAP:
// buf contain fio_send_request header and bitmap.
fio_send_pages_impl(fd[hdr.handle], out, buf, true);
break;
case FIO_SYNC:
/* open file and fsync it */
/* open file and fsync it */
tmp_fd = open(buf, O_WRONLY | PG_BINARY, FILE_PERMISSIONS);
if (tmp_fd < 0)
hdr.arg = errno;
@@ -1657,6 +1800,10 @@ void fio_communicate(int in, int out)
crc = pgFileGetCRC(buf, true, true);
IO_CHECK(fio_write_all(out, &crc, sizeof(crc)), sizeof(crc));
break;
case FIO_DISCONNECT:
hdr.cop = FIO_DISCONNECTED;
IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
break;
default:
Assert(false);
}
src/utils/file.h
+10 -7
View File
@@ -33,10 +33,18 @@ typedef enum
FIO_OPENDIR,
FIO_READDIR,
FIO_CLOSEDIR,
FIO_SEND_PAGES,
FIO_PAGE,
FIO_WRITE_COMPRESSED,
FIO_GET_CRC32
FIO_GET_CRC32,
/* used in fio_send_pages */
FIO_SEND_PAGES,
FIO_SEND_PAGES_PAGEMAP,
FIO_ERROR,
FIO_SEND_FILE_EOF,
FIO_SEND_FILE_CORRUPTION,
/* messages for closing connection */
FIO_DISCONNECT,
FIO_DISCONNECTED,
} fio_operations;
typedef enum
@@ -49,7 +57,6 @@ typedef enum
#define FIO_FDMAX 64
#define FIO_PIPE_MARKER 0x40000000
#define PAGE_CHECKSUM_MISMATCH (-256)
#define SYS_CHECK(cmd) do if ((cmd) < 0) { fprintf(stderr, "%s:%d: (%s) %s\n", __FILE__, __LINE__, #cmd, strerror(errno)); exit(EXIT_FAILURE); } while (0)
#define IO_CHECK(cmd, size) do { int _rc = (cmd); if (_rc != (size)) fio_error(_rc, size, __FILE__, __LINE__); } while (0)
@@ -83,10 +90,6 @@ extern int fio_fclose(FILE* f);
extern int fio_ffstat(FILE* f, struct stat* st);
extern void fio_error(int rc, int size, char const* file, int line);
struct pgFile;
extern int fio_send_pages(FILE* in, FILE* out, struct pgFile *file, XLogRecPtr horizonLsn,
BlockNumber* nBlocksSkipped, int calg, int clevel);
extern int fio_open(char const* name, int mode, fio_location location);
extern ssize_t fio_write(int fd, void const* buf, size_t size);
extern ssize_t fio_read(int fd, void* buf, size_t size);
tests/backup.py
+372 -74
View File
@@ -488,6 +488,7 @@ class BackupTest(ProbackupTest, unittest.TestCase):
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
if self.ptrack and node.major_version > 11:
@@ -499,18 +500,29 @@ class BackupTest(ProbackupTest, unittest.TestCase):
"postgres",
"create table t_heap as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,1000) i")
node.safe_psql(
"postgres",
"CHECKPOINT;")
"from generate_series(0,10000) i")
heap_path = node.safe_psql(
"postgres",
"select pg_relation_filepath('t_heap')").rstrip()
self.backup_node(
backup_dir, 'node', node,
backup_type="full", options=["-j", "4", "--stream"])
node.safe_psql(
"postgres",
"select count(*) from t_heap")
node.safe_psql(
"postgres",
"update t_heap set id = id + 10000")
node.stop()
with open(os.path.join(node.data_dir, heap_path), "rb+", 0) as f:
heap_fullpath = os.path.join(node.data_dir, heap_path)
with open(heap_fullpath, "rb+", 0) as f:
f.seek(9000)
f.write(b"bla")
f.flush()
@@ -518,6 +530,10 @@ class BackupTest(ProbackupTest, unittest.TestCase):
node.slow_start()
# self.backup_node(
# backup_dir, 'node', node,
# backup_type="full", options=["-j", "4", "--stream"])
try:
self.backup_node(
backup_dir, 'node', node,
@@ -525,11 +541,360 @@ class BackupTest(ProbackupTest, unittest.TestCase):
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because tablespace mapping is incorrect"
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
if self.ptrack:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page verification failed, calculated checksum'.format(
heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="delta", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page verification failed, calculated checksum'.format(
heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="page", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page verification failed, calculated checksum'.format(
heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
if self.ptrack:
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="ptrack", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertTrue(
'WARNING: page verification failed, '
'calculated checksum' in e.message and
'ERROR: query failed: ERROR: '
'invalid page in block 1 of relation' in e.message and
'ERROR: Data files transferring failed' 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)
# @unittest.skip("skip")
def test_backup_detect_invalid_block_header(self):
"""make node, corrupt some page, check that backup failed"""
fname = self.id().split('.')[3]
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
ptrack_enable=self.ptrack,
initdb_params=['--data-checksums'])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
if self.ptrack and node.major_version > 11:
node.safe_psql(
"postgres",
"create extension ptrack")
node.safe_psql(
"postgres",
"create table t_heap as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,10000) i")
heap_path = node.safe_psql(
"postgres",
"select pg_relation_filepath('t_heap')").rstrip()
self.backup_node(
backup_dir, 'node', node,
backup_type="full", options=["-j", "4", "--stream"])
node.safe_psql(
"postgres",
"select count(*) from t_heap")
node.safe_psql(
"postgres",
"update t_heap set id = id + 10000")
node.stop()
heap_fullpath = os.path.join(node.data_dir, heap_path)
with open(heap_fullpath, "rb+", 0) as f:
f.seek(8193)
f.write(b"blahblahblahblah")
f.flush()
f.close
node.slow_start()
# self.backup_node(
# backup_dir, 'node', node,
# backup_type="full", options=["-j", "4", "--stream"])
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="full", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page header invalid, pd_lower'.format(heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="delta", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page header invalid, pd_lower'.format(heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="page", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page header invalid, pd_lower'.format(heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
if self.ptrack:
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="ptrack", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertTrue(
'WARNING: page verification failed, '
'calculated checksum' in e.message and
'ERROR: query failed: ERROR: '
'invalid page in block 1 of relation' in e.message and
'ERROR: Data files transferring failed' 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)
# @unittest.skip("skip")
def test_backup_detect_missing_permissions(self):
"""make node, corrupt some page, check that backup failed"""
fname = self.id().split('.')[3]
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
ptrack_enable=self.ptrack,
initdb_params=['--data-checksums'])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
if self.ptrack and node.major_version > 11:
node.safe_psql(
"postgres",
"create extension ptrack")
node.safe_psql(
"postgres",
"create table t_heap as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,10000) i")
heap_path = node.safe_psql(
"postgres",
"select pg_relation_filepath('t_heap')").rstrip()
self.backup_node(
backup_dir, 'node', node,
backup_type="full", options=["-j", "4", "--stream"])
node.safe_psql(
"postgres",
"select count(*) from t_heap")
node.safe_psql(
"postgres",
"update t_heap set id = id + 10000")
node.stop()
heap_fullpath = os.path.join(node.data_dir, heap_path)
with open(heap_fullpath, "rb+", 0) as f:
f.seek(8193)
f.write(b"blahblahblahblah")
f.flush()
f.close
node.slow_start()
# self.backup_node(
# backup_dir, 'node', node,
# backup_type="full", options=["-j", "4", "--stream"])
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="full", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page header invalid, pd_lower'.format(heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="delta", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page header invalid, pd_lower'.format(heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="page", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Corruption detected in file "{0}", block 1: '
'page header invalid, pd_lower'.format(heap_fullpath),
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
sleep(1)
if self.ptrack:
try:
self.backup_node(
backup_dir, 'node', node,
backup_type="ptrack", options=["-j", "4", "--stream"])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of block corruption"
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertTrue(
'WARNING: page verification failed, '
'calculated checksum' in e.message and
@@ -538,24 +903,6 @@ class BackupTest(ProbackupTest, unittest.TestCase):
'ERROR: Data files transferring failed' in e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
else:
if self.remote:
self.assertTrue(
"ERROR: Failed to read file" in e.message and
"data file checksum mismatch" in e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
else:
self.assertIn(
'WARNING: Corruption detected in file',
e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
self.assertIn(
'ERROR: Data file corruption',
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)
@@ -2355,52 +2702,3 @@ class BackupTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_streaming_timeout(self):
"""
Illustrate the problem of loosing exact error
message because our WAL streaming engine is "borrowed"
from pg_receivexlog
"""
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={
'checkpoint_timeout': '1h',
'wal_sender_timeout': '5s'})
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
node.slow_start()
# FULL backup
gdb = self.backup_node(
backup_dir, 'node', node, gdb=True,
options=['--stream', '--log-level-file=LOG'])
gdb.set_breakpoint('pg_stop_backup')
gdb.run_until_break()
sleep(10)
gdb.continue_execution_until_error()
gdb._execute('detach')
sleep(2)
log_file_path = os.path.join(backup_dir, 'log', 'pg_probackup.log')
with open(log_file_path) as f:
log_content = f.read()
self.assertIn(
'could not receive data from WAL stream',
log_content)
self.assertIn(
'ERROR: Problem in receivexlog',
log_content)
# Clean after yourself
self.del_test_dir(module_name, fname)
tests/compatibility.py
+80
View File
@@ -765,3 +765,83 @@ class CompatibilityTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_page_vacuum_truncate_compression(self):
"""
make node, create table, take full backup,
delete all data, vacuum relation,
take page backup, insert some data,
take second page backup,
restore latest page backup using new binary
and check data correctness
old binary should be 2.2.x version
"""
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, old_binary=True)
self.add_instance(backup_dir, 'node', node, old_binary=True)
self.set_archiving(backup_dir, 'node', node, old_binary=True)
node.slow_start()
node.safe_psql(
"postgres",
"create sequence t_seq; "
"create table t_heap as select i as id, "
"md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,1024) i")
node.safe_psql(
"postgres",
"vacuum t_heap")
self.backup_node(
backup_dir, 'node',node, old_binary=True, options=['--compress'])
node.safe_psql(
"postgres",
"delete from t_heap")
node.safe_psql(
"postgres",
"vacuum t_heap")
self.backup_node(
backup_dir, 'node', node, backup_type='page',
old_binary=True, options=['--compress'])
node.safe_psql(
"postgres",
"insert into t_heap select i as id, "
"md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,1) i")
self.backup_node(
backup_dir, 'node', node, backup_type='page',
old_binary=True, options=['--compress'])
pgdata = self.pgdata_content(node.data_dir)
node_restored = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node_restored'))
node_restored.cleanup()
self.restore_node(backup_dir, 'node', node_restored)
# Physical comparison
pgdata_restored = self.pgdata_content(node_restored.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
self.set_auto_conf(node_restored, {'port': node_restored.port})
node_restored.slow_start()
# Clean after yourself
self.del_test_dir(module_name, fname)
tests/false_positive.py
+254
View File
@@ -294,3 +294,257 @@ class FalsePositive(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
@unittest.expectedFailure
def test_pg_10_waldir(self):
"""
test group access for PG >= 11
"""
if self.pg_config_version < self.version_to_num('10.0'):
return unittest.skip('You need PostgreSQL >= 10 for this test')
fname = self.id().split('.')[3]
wal_dir = os.path.join(
os.path.join(self.tmp_path, module_name, fname), 'wal_dir')
shutil.rmtree(wal_dir, ignore_errors=True)
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
initdb_params=[
'--data-checksums',
'--waldir={0}'.format(wal_dir)])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
node.slow_start()
# take FULL backup
self.backup_node(
backup_dir, 'node', node, options=['--stream'])
pgdata = self.pgdata_content(node.data_dir)
# restore backup
node_restored = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node_restored'))
node_restored.cleanup()
self.restore_node(
backup_dir, 'node', node_restored)
# compare pgdata permissions
pgdata_restored = self.pgdata_content(node_restored.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
self.assertTrue(
os.path.islink(os.path.join(node_restored.data_dir, 'pg_wal')),
'pg_wal should be symlink')
# Clean after yourself
self.del_test_dir(module_name, fname)
@unittest.expectedFailure
# @unittest.skip("skip")
def test_recovery_target_time_backup_victim(self):
"""
Check that for validation to recovery target
probackup chooses valid backup
https://github.com/postgrespro/pg_probackup/issues/104
"""
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'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
# FULL backup
self.backup_node(backup_dir, 'node', node)
node.safe_psql(
"postgres",
"create table t_heap as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,10000) i")
target_time = node.safe_psql(
"postgres",
"select now()").rstrip()
node.safe_psql(
"postgres",
"create table t_heap1 as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,100) i")
gdb = self.backup_node(backup_dir, 'node', node, gdb=True)
gdb.set_breakpoint('pg_stop_backup')
gdb.run_until_break()
gdb.remove_all_breakpoints()
gdb._execute('signal SIGINT')
gdb.continue_execution_until_error()
backup_id = self.show_pb(backup_dir, 'node')[1]['id']
self.assertEqual(
'ERROR',
self.show_pb(backup_dir, 'node', backup_id)['status'],
'Backup STATUS should be "ERROR"')
self.validate_pb(
backup_dir, 'node',
options=['--recovery-target-time={0}'.format(target_time)])
# Clean after yourself
self.del_test_dir(module_name, fname)
@unittest.expectedFailure
# @unittest.skip("skip")
def test_recovery_target_lsn_backup_victim(self):
"""
Check that for validation to recovery target
probackup chooses valid backup
https://github.com/postgrespro/pg_probackup/issues/104
"""
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'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
# FULL backup
self.backup_node(backup_dir, 'node', node)
node.safe_psql(
"postgres",
"create table t_heap as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,10000) i")
node.safe_psql(
"postgres",
"create table t_heap1 as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,100) i")
gdb = self.backup_node(
backup_dir, 'node', node,
options=['--log-level-console=LOG'], gdb=True)
gdb.set_breakpoint('pg_stop_backup')
gdb.run_until_break()
gdb.remove_all_breakpoints()
gdb._execute('signal SIGINT')
gdb.continue_execution_until_error()
backup_id = self.show_pb(backup_dir, 'node')[1]['id']
self.assertEqual(
'ERROR',
self.show_pb(backup_dir, 'node', backup_id)['status'],
'Backup STATUS should be "ERROR"')
self.switch_wal_segment(node)
target_lsn = self.show_pb(backup_dir, 'node', backup_id)['start-lsn']
self.validate_pb(
backup_dir, 'node',
options=['--recovery-target-lsn={0}'.format(target_lsn)])
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
@unittest.expectedFailure
def test_streaming_timeout(self):
"""
Illustrate the problem of loosing exact error
message because our WAL streaming engine is "borrowed"
from pg_receivexlog
"""
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={
'checkpoint_timeout': '1h',
'wal_sender_timeout': '5s'})
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
node.slow_start()
# FULL backup
gdb = self.backup_node(
backup_dir, 'node', node, gdb=True,
options=['--stream', '--log-level-file=LOG'])
gdb.set_breakpoint('pg_stop_backup')
gdb.run_until_break()
sleep(10)
gdb.continue_execution_until_error()
gdb._execute('detach')
sleep(2)
log_file_path = os.path.join(backup_dir, 'log', 'pg_probackup.log')
with open(log_file_path) as f:
log_content = f.read()
self.assertIn(
'could not receive data from WAL stream',
log_content)
self.assertIn(
'ERROR: Problem in receivexlog',
log_content)
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
@unittest.expectedFailure
def test_validate_all_empty_catalog(self):
"""
"""
fname = self.id().split('.')[3]
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
initdb_params=['--data-checksums'])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
try:
self.validate_pb(backup_dir)
self.assertEqual(
1, 0,
"Expecting Error because backup_dir is empty.\n "
"Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: This backup catalog contains no backup instances',
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)
tests/page.py
+7 -13
View File
@@ -391,13 +391,12 @@ class PageTest(ProbackupTest, unittest.TestCase):
# PGBENCH STUFF
pgbench = node.pgbench(options=['-T', '50', '-c', '1', '--no-vacuum'])
pgbench.wait()
node.safe_psql("postgres", "checkpoint")
# GET LOGICAL CONTENT FROM NODE
result = node.safe_psql("postgres", "select * from pgbench_accounts")
# PAGE BACKUP
self.backup_node(
backup_dir, 'node', node, backup_type='page')
self.backup_node(backup_dir, 'node', node, backup_type='page')
# GET PHYSICAL CONTENT FROM NODE
pgdata = self.pgdata_content(node.data_dir)
@@ -464,18 +463,15 @@ class PageTest(ProbackupTest, unittest.TestCase):
"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"
)
" from generate_series(0,100) i")
node.safe_psql(
"postgres",
"delete from t_heap"
)
"delete from t_heap")
node.safe_psql(
"postgres",
"vacuum t_heap"
)
"vacuum t_heap")
# PAGE BACKUP
self.backup_node(
@@ -485,8 +481,7 @@ class PageTest(ProbackupTest, unittest.TestCase):
# RESTORE
node_restored = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node_restored')
)
base_dir=os.path.join(module_name, fname, 'node_restored'))
node_restored.cleanup()
self.restore_node(
@@ -1122,8 +1117,7 @@ class PageTest(ProbackupTest, unittest.TestCase):
# RESTORE
node_restored = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node_restored')
)
base_dir=os.path.join(module_name, fname, 'node_restored'))
node_restored.cleanup()
self.restore_node(
tests/pgpro560.py
+1 -1
View File
@@ -43,7 +43,7 @@ class CheckSystemID(ProbackupTest, unittest.TestCase):
"Output: {0} \n CMD: {1}".format(repr(self.output), self.cmd))
except ProbackupException as e:
self.assertTrue(
'ERROR: could not open file' in e.message and
'ERROR: Could not open file' in e.message and
'pg_control' in e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
tests/remote.py
+21 -16
View File
@@ -24,22 +24,27 @@ class RemoteTest(ProbackupTest, unittest.TestCase):
self.add_instance(backup_dir, 'node', node)
node.slow_start()
try:
self.backup_node(
backup_dir, 'node',
node, options=['--remote-proto=ssh', '--stream'], no_remote=True)
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because remote-host option is missing."
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
"Insert correct error",
e.message,
"\n Unexpected Error Message: {0}\n CMD: {1}".format(
repr(e.message), self.cmd))
output = self.backup_node(
backup_dir, 'node', node,
options=['--stream'], no_remote=True, return_id=False)
self.assertIn('remote: false', output)
# try:
# self.backup_node(
# backup_dir, 'node',
# node, options=['--remote-proto=ssh', '--stream'], no_remote=True)
# # we should die here because exception is what we expect to happen
# self.assertEqual(
# 1, 0,
# "Expecting Error because remote-host option is missing."
# "\n Output: {0} \n CMD: {1}".format(
# repr(self.output), self.cmd))
# except ProbackupException as e:
# self.assertIn(
# "Insert correct error",
# 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)
tests/restore.py
+60 -64
View File
@@ -1915,8 +1915,7 @@ class RestoreTest(ProbackupTest, unittest.TestCase):
node.slow_start()
# Take FULL
self.backup_node(
backup_dir, 'node', node)
self.backup_node(backup_dir, 'node', node)
if self.get_version(node) >= self.version_to_num('12.0'):
recovery_conf = os.path.join(node.data_dir, 'probackup_recovery.conf')
@@ -1925,27 +1924,28 @@ class RestoreTest(ProbackupTest, unittest.TestCase):
# restore
node.cleanup()
self.restore_node(
backup_dir, 'node', node)
self.restore_node(backup_dir, 'node', node)
# with open(recovery_conf, 'r') as f:
# print(f.read())
# hash_1 = hashlib.md5(
# open(recovery_conf, 'rb').read()).hexdigest()
hash_1 = hashlib.md5(
open(recovery_conf, 'rb').read()).hexdigest()
with open(recovery_conf, 'r') as f:
content_1 = f.read()
# restore
node.cleanup()
self.restore_node(
backup_dir, 'node', node, options=['--recovery-target=latest'])
# with open(recovery_conf, 'r') as f:
# print(f.read())
self.restore_node(backup_dir, 'node', node, options=['--recovery-target=latest'])
hash_2 = hashlib.md5(
open(recovery_conf, 'rb').read()).hexdigest()
# hash_2 = hashlib.md5(
# open(recovery_conf, 'rb').read()).hexdigest()
self.assertEqual(hash_1, hash_2)
with open(recovery_conf, 'r') as f:
content_2 = f.read()
self.assertEqual(content_1, content_2)
# self.assertEqual(hash_1, hash_2)
# Clean after yourself
self.del_test_dir(module_name, fname)
@@ -2231,55 +2231,6 @@ class RestoreTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_pg_10_waldir(self):
"""
test group access for PG >= 11
"""
if self.pg_config_version < self.version_to_num('10.0'):
return unittest.skip('You need PostgreSQL >= 10 for this test')
fname = self.id().split('.')[3]
wal_dir = os.path.join(
os.path.join(self.tmp_path, module_name, fname), 'wal_dir')
shutil.rmtree(wal_dir, ignore_errors=True)
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
initdb_params=[
'--data-checksums',
'--waldir={0}'.format(wal_dir)])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
node.slow_start()
# take FULL backup
self.backup_node(
backup_dir, 'node', node, options=['--stream'])
pgdata = self.pgdata_content(node.data_dir)
# restore backup
node_restored = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node_restored'))
node_restored.cleanup()
self.restore_node(
backup_dir, 'node', node_restored)
# compare pgdata permissions
pgdata_restored = self.pgdata_content(node_restored.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
self.assertTrue(
os.path.islink(os.path.join(node_restored.data_dir, 'pg_wal')),
'pg_wal should be symlink')
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_restore_concurrent_drop_table(self):
""""""
@@ -3427,3 +3378,48 @@ class RestoreTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_restore_primary_slot_info(self):
"""
"""
fname = self.id().split('.')[3]
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
set_replication=True,
initdb_params=['--data-checksums'])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
node.slow_start()
# Take FULL
self.backup_node(backup_dir, 'node', node, options=['--stream'])
node.pgbench_init(scale=1)
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
node.safe_psql(
"SELECT pg_create_physical_replication_slot('master_slot')")
self.restore_node(
backup_dir, 'node', replica,
options=['-R', '--primary-slot-name=master_slot'])
self.set_auto_conf(replica, {'port': replica.port})
self.set_auto_conf(replica, {'hot_standby': 'on'})
if self.get_version(node) >= self.version_to_num('12.0'):
standby_signal = os.path.join(replica.data_dir, 'standby.signal')
self.assertTrue(
os.path.isfile(standby_signal),
"File '{0}' do not exists".format(standby_signal))
replica.slow_start(replica=True)
# Clean after yourself
self.del_test_dir(module_name, fname)
tests/validate.py
-154
View File
@@ -13,36 +13,6 @@ module_name = 'validate'
class ValidateTest(ProbackupTest, unittest.TestCase):
# @unittest.skip("skip")
# @unittest.expectedFailure
def test_validate_all_empty_catalog(self):
"""
"""
fname = self.id().split('.')[3]
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
initdb_params=['--data-checksums'])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
try:
self.validate_pb(backup_dir)
self.assertEqual(
1, 0,
"Expecting Error because backup_dir is empty.\n "
"Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: This backup catalog contains no backup instances',
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)
# @unittest.skip("skip")
# @unittest.expectedFailure
def test_basic_validate_nullified_heap_page_backup(self):
@@ -3538,130 +3508,6 @@ class ValidateTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.expectedFailure
# @unittest.skip("skip")
def test_recovery_target_time_backup_victim(self):
"""
Check that for validation to recovery target
probackup chooses valid backup
https://github.com/postgrespro/pg_probackup/issues/104
"""
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'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
# FULL backup
self.backup_node(backup_dir, 'node', node)
node.safe_psql(
"postgres",
"create table t_heap as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,10000) i")
target_time = node.safe_psql(
"postgres",
"select now()").rstrip()
node.safe_psql(
"postgres",
"create table t_heap1 as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,100) i")
gdb = self.backup_node(backup_dir, 'node', node, gdb=True)
gdb.set_breakpoint('pg_stop_backup')
gdb.run_until_break()
gdb.remove_all_breakpoints()
gdb._execute('signal SIGINT')
gdb.continue_execution_until_error()
backup_id = self.show_pb(backup_dir, 'node')[1]['id']
self.assertEqual(
'ERROR',
self.show_pb(backup_dir, 'node', backup_id)['status'],
'Backup STATUS should be "ERROR"')
self.validate_pb(
backup_dir, 'node',
options=['--recovery-target-time={0}'.format(target_time)])
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.expectedFailure
# @unittest.skip("skip")
def test_recovery_target_lsn_backup_victim(self):
"""
Check that for validation to recovery target
probackup chooses valid backup
https://github.com/postgrespro/pg_probackup/issues/104
"""
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'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
# FULL backup
self.backup_node(backup_dir, 'node', node)
node.safe_psql(
"postgres",
"create table t_heap as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,10000) i")
node.safe_psql(
"postgres",
"create table t_heap1 as select 1 as id, md5(i::text) as text, "
"md5(repeat(i::text,10))::tsvector as tsvector "
"from generate_series(0,100) i")
gdb = self.backup_node(
backup_dir, 'node', node,
options=['--log-level-console=LOG'], gdb=True)
gdb.set_breakpoint('pg_stop_backup')
gdb.run_until_break()
gdb.remove_all_breakpoints()
gdb._execute('signal SIGINT')
gdb.continue_execution_until_error()
backup_id = self.show_pb(backup_dir, 'node')[1]['id']
self.assertEqual(
'ERROR',
self.show_pb(backup_dir, 'node', backup_id)['status'],
'Backup STATUS should be "ERROR"')
self.switch_wal_segment(node)
target_lsn = self.show_pb(backup_dir, 'node', backup_id)['start-lsn']
self.validate_pb(
backup_dir, 'node',
options=['--recovery-target-lsn={0}'.format(target_lsn)])
# Clean after yourself
self.del_test_dir(module_name, fname)
@unittest.skip("skip")
def test_partial_validate_empty_and_mangled_database_map(self):
"""