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 #143] Multi-timeline incremental chain

Browse Source
This commit is contained in:
Grigory Smolkin
2020-04-03 18:08:53 +03:00
parent a196073944
commit 4a94fdaa33
11 changed files with 1106 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/pgprobackup.xml
+4 -11
View File
@@ -426,14 +426,6 @@ doc/src/sgml/pgprobackup.sgml
or <application>libc</application>/<application>libicu</application> versions.
</para>
</listitem>
<listitem>
<para>
All backups in the incremental chain must belong to the same
timeline. For example, if you have taken incremental backups on a
standby server that gets promoted, you have to take another FULL
backup.
</para>
</listitem>
</itemizedlist>
</para>
</refsect2>
@@ -753,9 +745,10 @@ ALTER ROLE backup WITH REPLICATION;
<title>Setting up Continuous WAL Archiving</title>
<para>
Making backups in PAGE backup mode, performing
<link linkend="pbk-performing-point-in-time-pitr-recovery">PITR</link>
and making backups with
<link linkend="pbk-archive-mode">ARCHIVE</link> WAL delivery mode
<link linkend="pbk-performing-point-in-time-pitr-recovery">PITR</link>,
making backups with
<link linkend="pbk-archive-mode">ARCHIVE</link> WAL delivery mode and
running incremental backup after timeline switch
require
<ulink url="https://postgrespro.com/docs/postgresql/current/continuous-archiving.html">continuous
WAL archiving</ulink> to be enabled. To set up continuous
src/backup.c
+63 -9
View File
@@ -153,6 +153,10 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
PGconn *master_conn = NULL;
PGconn *pg_startbackup_conn = NULL;
/* used for multitimeline incremental backup */
parray *tli_list = NULL;
/* for fancy reporting */
time_t start_time, end_time;
char pretty_time[20];
@@ -181,17 +185,43 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
current.backup_mode == BACKUP_MODE_DIFF_PTRACK ||
current.backup_mode == BACKUP_MODE_DIFF_DELTA)
{
char prev_backup_filelist_path[MAXPGPATH];
/* get list of backups already taken */
backup_list = catalog_get_backup_list(instance_name, INVALID_BACKUP_ID);
prev_backup = catalog_get_last_data_backup(backup_list, current.tli, current.start_time);
if (prev_backup == NULL)
elog(ERROR, "Valid backup on current timeline %X is not found. "
"Create new FULL backup before an incremental one.",
{
/* try to setup multi-timeline backup chain */
elog(WARNING, "Valid backup on current timeline %u is not found, "
"try to look up on previous timelines",
current.tli);
tli_list = catalog_get_timelines(&instance_config);
if (parray_num(tli_list) == 0)
elog(WARNING, "Cannot find valid backup on previous timelines, "
"WAL archive is not available");
else
{
prev_backup = get_multi_timeline_parent(backup_list, tli_list, current.tli,
current.start_time, &instance_config);
if (prev_backup == NULL)
elog(WARNING, "Cannot find valid backup on previous timelines");
}
/* failed to find suitable parent, error out */
if (!prev_backup)
elog(ERROR, "Create new full backup before an incremental one");
}
}
if (prev_backup)
{
char prev_backup_filelist_path[MAXPGPATH];
elog(INFO, "Parent backup: %s", base36enc(prev_backup->start_time));
join_path_components(prev_backup_filelist_path, prev_backup->root_dir,
DATABASE_FILE_LIST);
/* Files of previous backup needed by DELTA backup */
@@ -378,8 +408,10 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
if (current.backup_mode == BACKUP_MODE_DIFF_PAGE ||
current.backup_mode == BACKUP_MODE_DIFF_PTRACK)
{
elog(INFO, "Compiling pagemap of changed blocks");
bool pagemap_isok = true;
time(&start_time);
elog(INFO, "Extracting pagemap of changed blocks");
if (current.backup_mode == BACKUP_MODE_DIFF_PAGE)
{
@@ -388,8 +420,9 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
* reading WAL segments present in archives up to the point
* where this backup has started.
*/
extractPageMap(arclog_path, current.tli, instance_config.xlog_seg_size,
prev_backup->start_lsn, current.start_lsn);
pagemap_isok = extractPageMap(arclog_path, instance_config.xlog_seg_size,
prev_backup->start_lsn, prev_backup->tli,
current.start_lsn, current.tli, tli_list);
}
else if (current.backup_mode == BACKUP_MODE_DIFF_PTRACK)
{
@@ -407,8 +440,14 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
}
time(&end_time);
elog(INFO, "Pagemap compiled, time elapsed %.0f sec",
difftime(end_time, start_time));
/* TODO: add ms precision */
if (pagemap_isok)
elog(INFO, "Pagemap successfully extracted, time elapsed %.0f sec",
difftime(end_time, start_time));
else
elog(ERROR, "Pagemap extraction failed, time elasped: %.0f sec",
difftime(end_time, start_time));
}
/*
@@ -667,6 +706,15 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
elog(INFO, "Backup files are synced, time elapsed: %s", pretty_time);
}
/* be paranoid about instance been from the past */
if (current.backup_mode != BACKUP_MODE_FULL &&
current.stop_lsn < prev_backup->stop_lsn)
elog(ERROR, "Current backup STOP LSN %X/%X is lower than STOP LSN %X/%X of previous backup %s. "
"It may indicate that we are trying to backup PostgreSQL instance from the past.",
(uint32) (current.stop_lsn >> 32), (uint32) (current.stop_lsn),
(uint32) (prev_backup->stop_lsn >> 32), (uint32) (prev_backup->stop_lsn),
base36enc(prev_backup->stop_lsn));
/* clean external directories list */
if (external_dirs)
free_dir_list(external_dirs);
@@ -678,6 +726,12 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync)
parray_free(backup_list);
}
if (tli_list)
{
parray_walk(tli_list, timelineInfoFree);
parray_free(tli_list);
}
parray_walk(backup_files_list, pgFileFree);
parray_free(backup_files_list);
backup_files_list = NULL;
src/catalog.c
+161 -4
View File
@@ -42,6 +42,24 @@ timelineInfoNew(TimeLineID tli)
return tlinfo;
}
/* free timelineInfo object */
void
timelineInfoFree(void *tliInfo)
{
timelineInfo *tli = (timelineInfo *) tliInfo;
parray_walk(tli->xlog_filelist, pgFileFree);
parray_free(tli->xlog_filelist);
if (tli->backups)
{
parray_walk(tli->backups, pgBackupFree);
parray_free(tli->backups);
}
pfree(tliInfo);
}
/* Iterate over locked backups and delete locks files */
static void
unlink_lock_atexit(void)
@@ -621,11 +639,7 @@ catalog_get_last_data_backup(parray *backup_list, TimeLineID tli, time_t current
* anomalies.
*/
if (is_parent(full_backup->start_time, backup, true))
{
elog(INFO, "Parent backup: %s",
base36enc(backup->start_time));
return backup;
}
}
}
/* skip yourself */
@@ -641,6 +655,149 @@ catalog_get_last_data_backup(parray *backup_list, TimeLineID tli, time_t current
return NULL;
}
/*
* For multi-timeline chain, look up suitable parent for incremental backup.
* Multi-timeline chain has full backup and one or more descendants located
* on different timelines.
*/
pgBackup *
get_multi_timeline_parent(parray *backup_list, parray *tli_list,
TimeLineID current_tli, time_t current_start_time,
InstanceConfig *instance)
{
int i;
timelineInfo *my_tlinfo = NULL;
timelineInfo *tmp_tlinfo = NULL;
pgBackup *ancestor_backup = NULL;
/* there are no timelines in the archive */
if (parray_num(tli_list) == 0)
return NULL;
/* look for current timelineInfo */
for (i = 0; i < parray_num(tli_list); i++)
{
timelineInfo *tlinfo = (timelineInfo *) parray_get(tli_list, i);
if (tlinfo->tli == current_tli)
{
my_tlinfo = tlinfo;
break;
}
}
if (my_tlinfo == NULL)
return NULL;
/* Locate tlinfo of suitable full backup.
* Consider this example:
* t3 s2-------X <-! We are here
* /
* t2 s1----D---*----E--->
* /
* t1--A--B--*---C------->
*
* A, E - full backups
* B, C, D - incremental backups
*
* We must find A.
*/
tmp_tlinfo = my_tlinfo;
while (tmp_tlinfo->parent_link)
{
/* if timeline has backups, iterate over them */
if (tmp_tlinfo->parent_link->backups)
{
for (i = 0; i < parray_num(tmp_tlinfo->parent_link->backups); i++)
{
pgBackup *backup = (pgBackup *) parray_get(tmp_tlinfo->parent_link->backups, i);
if (backup->backup_mode == BACKUP_MODE_FULL &&
(backup->status == BACKUP_STATUS_OK ||
backup->status == BACKUP_STATUS_DONE) &&
backup->stop_lsn <= tmp_tlinfo->switchpoint)
{
ancestor_backup = backup;
break;
}
}
}
if (ancestor_backup)
break;
tmp_tlinfo = tmp_tlinfo->parent_link;
}
/* failed to find valid FULL backup on parent timelines */
if (!ancestor_backup)
return NULL;
else
elog(LOG, "Latest valid full backup: %s, tli: %i",
base36enc(ancestor_backup->start_time), ancestor_backup->tli);
/* At this point we found suitable full backup,
* now we must find his latest child, suitable to be
* parent of current incremental backup.
* Consider this example:
* t3 s2-------X <-! We are here
* /
* t2 s1----D---*----E--->
* /
* t1--A--B--*---C------->
*
* A, E - full backups
* B, C, D - incremental backups
*
* We found A, now we must find D.
*/
/* Optimistically, look on current timeline for valid incremental backup, child of ancestor */
if (my_tlinfo->backups)
{
for (i = 0; i < parray_num(my_tlinfo->backups); i++)
{
pgBackup *tmp_backup = NULL;
pgBackup *backup = (pgBackup *) parray_get(my_tlinfo->backups, i);
/* found suitable parent */
if (scan_parent_chain(backup, &tmp_backup) == 2 &&
is_parent(ancestor_backup->start_time, backup, false))
return backup;
}
}
/* Iterate over parent timelines and look for a valid backup, child of ancestor */
tmp_tlinfo = my_tlinfo;
while (tmp_tlinfo->parent_link)
{
/* if timeline has backups, iterate over them */
if (tmp_tlinfo->parent_link->backups)
{
for (i = 0; i < parray_num(tmp_tlinfo->parent_link->backups); i++)
{
pgBackup *tmp_backup = NULL;
pgBackup *backup = (pgBackup *) parray_get(tmp_tlinfo->parent_link->backups, i);
/* We are not interested in backups
* located outside of our timeline history
*/
if (backup->stop_lsn > tmp_tlinfo->switchpoint)
continue;
if (scan_parent_chain(backup, &tmp_backup) == 2 &&
is_parent(ancestor_backup->start_time, backup, true))
return backup;
}
}
tmp_tlinfo = tmp_tlinfo->parent_link;
}
return NULL;
}
/* create backup directory in $BACKUP_PATH */
int
pgBackupCreateDir(pgBackup *backup)
src/parsexlog.c
+171 -35
View File
@@ -138,6 +138,9 @@ typedef struct
*/
bool got_target;
/* Should we read record, located at endpoint position */
bool inclusive_endpoint;
/*
* Return value from the thread.
* 0 means there is no error, 1 - there is an error.
@@ -162,7 +165,8 @@ static bool RunXLogThreads(const char *archivedir,
XLogRecPtr startpoint, XLogRecPtr endpoint,
bool consistent_read,
xlog_record_function process_record,
XLogRecTarget *last_rec);
XLogRecTarget *last_rec,
bool inclusive_endpoint);
//static XLogReaderState *InitXLogThreadRead(xlog_thread_arg *arg);
static bool SwitchThreadToNextWal(XLogReaderState *xlogreader,
xlog_thread_arg *arg);
@@ -231,18 +235,118 @@ static XLogRecPtr wal_target_lsn = InvalidXLogRecPtr;
* Pagemap extracting is processed using threads. Each thread reads single WAL
* file.
*/
void
extractPageMap(const char *archivedir, TimeLineID tli, uint32 wal_seg_size,
XLogRecPtr startpoint, XLogRecPtr endpoint)
bool
extractPageMap(const char *archivedir, uint32 wal_seg_size,
XLogRecPtr startpoint, TimeLineID start_tli,
XLogRecPtr endpoint, TimeLineID end_tli,
parray *tli_list)
{
bool extract_isok = true;
bool extract_isok = false;
extract_isok = RunXLogThreads(archivedir, 0, InvalidTransactionId,
InvalidXLogRecPtr, tli, wal_seg_size,
startpoint, endpoint, false, extractPageInfo,
NULL);
if (!extract_isok)
elog(ERROR, "Pagemap compiling failed");
if (start_tli == end_tli)
/* easy case */
extract_isok = RunXLogThreads(archivedir, 0, InvalidTransactionId,
InvalidXLogRecPtr, end_tli, wal_seg_size,
startpoint, endpoint, false, extractPageInfo,
NULL, true);
else
{
/* We have to process WAL located on several different xlog intervals,
* located on different timelines.
*
* Consider this example:
* t3 C-----X <!- We are here
* /
* t2 B---*-->
* /
* t1 -A----*------->
*
* A - prev backup START_LSN
* B - switchpoint for t2, available as t2->switchpoint
* C - switch for t3, available as t3->switchpoint
* X - current backup START_LSN
*
* Intervals to be parsed:
* - [A,B) on t1
* - [B,C) on t2
* - [C,X] on t3
*/
int i;
parray *interval_list = parray_new();
timelineInfo *end_tlinfo = NULL;
timelineInfo *tmp_tlinfo = NULL;
XLogRecPtr prev_switchpoint = InvalidXLogRecPtr;
lsnInterval *wal_interval = NULL;
/* We must find TLI information about final timeline (t3 in example) */
for (i = 0; i < parray_num(tli_list); i++)
{
tmp_tlinfo = parray_get(tli_list, i);
if (tmp_tlinfo->tli == end_tli)
{
end_tlinfo = tmp_tlinfo;
break;
}
}
/* Iterate over timelines backward,
* starting with end_tli and ending with start_tli.
* For every timeline calculate LSN-interval that must be parsed.
*/
tmp_tlinfo = end_tlinfo;
while (tmp_tlinfo)
{
wal_interval = pgut_malloc(sizeof(lsnInterval));
wal_interval->tli = tmp_tlinfo->tli;
if (tmp_tlinfo->tli == end_tli)
{
wal_interval->begin_lsn = tmp_tlinfo->switchpoint;
wal_interval->end_lsn = endpoint;
}
else if (tmp_tlinfo->tli == start_tli)
{
wal_interval->begin_lsn = startpoint;
wal_interval->end_lsn = prev_switchpoint;
}
else
{
wal_interval->begin_lsn = tmp_tlinfo->switchpoint;
wal_interval->end_lsn = prev_switchpoint;
}
prev_switchpoint = tmp_tlinfo->switchpoint;
tmp_tlinfo = tmp_tlinfo->parent_link;
parray_append(interval_list, wal_interval);
}
for (i = parray_num(interval_list) - 1; i >= 0; i--)
{
bool inclusive_endpoint = false;
wal_interval = parray_get(interval_list, i);
/* In case of replica promotion, endpoints of intermediate
* timelines can be unreachable.
*/
if (wal_interval->tli == end_tli)
inclusive_endpoint = true;
extract_isok = RunXLogThreads(archivedir, 0, InvalidTransactionId,
InvalidXLogRecPtr, wal_interval->tli, wal_seg_size,
wal_interval->begin_lsn, wal_interval->end_lsn,
false, extractPageInfo, NULL, inclusive_endpoint);
if (!extract_isok)
break;
pg_free(wal_interval);
}
pg_free(interval_list);
}
return extract_isok;
}
/*
@@ -262,7 +366,7 @@ validate_backup_wal_from_start_to_stop(pgBackup *backup,
got_endpoint = RunXLogThreads(archivedir, 0, InvalidTransactionId,
InvalidXLogRecPtr, tli, xlog_seg_size,
backup->start_lsn, backup->stop_lsn,
false, NULL, NULL);
false, NULL, NULL, true);
if (!got_endpoint)
{
@@ -373,7 +477,7 @@ validate_wal(pgBackup *backup, const char *archivedir,
all_wal = all_wal ||
RunXLogThreads(archivedir, target_time, target_xid, target_lsn,
tli, wal_seg_size, backup->stop_lsn,
InvalidXLogRecPtr, true, validateXLogRecord, &last_rec);
InvalidXLogRecPtr, true, validateXLogRecord, &last_rec, true);
if (last_rec.rec_time > 0)
time2iso(last_timestamp, lengthof(last_timestamp),
timestamptz_to_time_t(last_rec.rec_time));
@@ -753,11 +857,26 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
if (!reader_data->xlogexists)
{
char xlogfname[MAXFNAMELEN];
char partial_file[MAXPGPATH];
GetXLogFileName(xlogfname, reader_data->tli, reader_data->xlogsegno,
wal_seg_size);
snprintf(reader_data->xlogpath, MAXPGPATH, "%s/%s", wal_archivedir,
xlogfname);
snprintf(partial_file, MAXPGPATH, "%s/%s.partial", wal_archivedir,
xlogfname);
snprintf(reader_data->gz_xlogpath, sizeof(reader_data->gz_xlogpath),
"%s.gz", reader_data->xlogpath);
/* 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))
{
snprintf(reader_data->xlogpath, MAXPGPATH, "%s/%s.partial", wal_archivedir,
xlogfname);
strncpy(reader_data->xlogpath, partial_file, MAXPGPATH);
}
if (fileExists(reader_data->xlogpath, FIO_BACKUP_HOST))
{
@@ -778,29 +897,23 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
}
#ifdef HAVE_LIBZ
/* Try to open compressed WAL segment */
else
else if (fileExists(reader_data->gz_xlogpath, FIO_BACKUP_HOST))
{
snprintf(reader_data->gz_xlogpath, sizeof(reader_data->gz_xlogpath),
"%s.gz", reader_data->xlogpath);
if (fileExists(reader_data->gz_xlogpath, FIO_BACKUP_HOST))
{
elog(LOG, "Thread [%d]: Opening compressed WAL segment \"%s\"",
reader_data->thread_num, reader_data->gz_xlogpath);
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,
reader_data->xlogexists = true;
reader_data->gz_xlogfile = fio_gzopen(reader_data->gz_xlogpath,
"rb", -1, FIO_BACKUP_HOST);
if (reader_data->gz_xlogfile == NULL)
{
elog(WARNING, "Thread [%d]: Could not open compressed WAL segment \"%s\": %s",
reader_data->thread_num, reader_data->gz_xlogpath,
strerror(errno));
return -1;
}
if (reader_data->gz_xlogfile == NULL)
{
elog(WARNING, "Thread [%d]: Could not open compressed WAL segment \"%s\": %s",
reader_data->thread_num, reader_data->gz_xlogpath,
strerror(errno));
return -1;
}
}
#endif
/* Exit without error if WAL segment doesn't exist */
if (!reader_data->xlogexists)
return -1;
@@ -923,7 +1036,7 @@ RunXLogThreads(const char *archivedir, time_t target_time,
TransactionId target_xid, XLogRecPtr target_lsn, TimeLineID tli,
uint32 segment_size, XLogRecPtr startpoint, XLogRecPtr endpoint,
bool consistent_read, xlog_record_function process_record,
XLogRecTarget *last_rec)
XLogRecTarget *last_rec, bool inclusive_endpoint)
{
pthread_t *threads;
xlog_thread_arg *thread_args;
@@ -932,17 +1045,27 @@ RunXLogThreads(const char *archivedir, time_t target_time,
XLogSegNo endSegNo = 0;
bool result = true;
if (!XRecOffIsValid(startpoint))
if (!XRecOffIsValid(startpoint) &&
!XRecOffIsNull(startpoint))
{
elog(ERROR, "Invalid startpoint value %X/%X",
(uint32) (startpoint >> 32), (uint32) (startpoint));
}
if (!XLogRecPtrIsInvalid(endpoint))
{
if (!XRecOffIsValid(endpoint))
if (XRecOffIsNull(endpoint))
{
GetXLogSegNo(endpoint, endSegNo, segment_size);
endSegNo--;
}
else if (!XRecOffIsValid(endpoint))
{
elog(ERROR, "Invalid endpoint value %X/%X",
(uint32) (endpoint >> 32), (uint32) (endpoint));
GetXLogSegNo(endpoint, endSegNo, segment_size);
}
else
GetXLogSegNo(endpoint, endSegNo, segment_size);
}
/* Initialize static variables for workers */
@@ -977,6 +1100,7 @@ RunXLogThreads(const char *archivedir, time_t target_time,
arg->startpoint = startpoint;
arg->endpoint = endpoint;
arg->endSegNo = endSegNo;
arg->inclusive_endpoint = inclusive_endpoint;
arg->got_target = false;
/* By default there is some error */
arg->ret = 1;
@@ -1192,6 +1316,18 @@ XLogThreadWorker(void *arg)
reader_data->thread_num,
(uint32) (errptr >> 32), (uint32) (errptr));
/* In we failed to read record located at endpoint position,
* and endpoint is not inclusive, do not consider this as an error.
*/
if (!thread_arg->inclusive_endpoint &&
errptr == thread_arg->endpoint)
{
elog(LOG, "Thread [%d]: Endpoint %X/%X is not inclusive, switch to the next timeline",
reader_data->thread_num,
(uint32) (thread_arg->endpoint >> 32), (uint32) (thread_arg->endpoint));
break;
}
/*
* If we don't have all WAL files from prev backup start_lsn to current
* start_lsn, we won't be able to build page map and PAGE backup will
src/pg_probackup.h
+16 -4
View File
@@ -474,7 +474,7 @@ struct timelineInfo {
TimeLineID tli; /* this timeline */
TimeLineID parent_tli; /* parent timeline. 0 if none */
timelineInfo *parent_link; /* link to parent timeline */
XLogRecPtr switchpoint; /* if this timeline has a parent
XLogRecPtr switchpoint; /* if this timeline has a parent, then
* switchpoint contains switchpoint LSN,
* otherwise 0 */
XLogSegNo begin_segno; /* first present segment in this timeline */
@@ -500,6 +500,13 @@ typedef struct xlogInterval
XLogSegNo end_segno;
} xlogInterval;
typedef struct lsnInterval
{
TimeLineID tli;
XLogRecPtr begin_lsn;
XLogRecPtr end_lsn;
} lsnInterval;
typedef enum xlogFileType
{
SEGMENT,
@@ -763,6 +770,10 @@ extern void catalog_lock_backup_list(parray *backup_list, int from_idx,
extern pgBackup *catalog_get_last_data_backup(parray *backup_list,
TimeLineID tli,
time_t current_start_time);
extern pgBackup *get_multi_timeline_parent(parray *backup_list, parray *tli_list,
TimeLineID current_tli, time_t current_start_time,
InstanceConfig *instance);
extern void timelineInfoFree(void *tliInfo);
extern parray *catalog_get_timelines(InstanceConfig *instance);
extern void do_set_backup(const char *instance_name, time_t backup_id,
pgSetBackupParams *set_backup_params);
@@ -898,9 +909,10 @@ extern bool create_empty_file(fio_location from_location, const char *to_root,
extern bool check_file_pages(pgFile *file, XLogRecPtr stop_lsn,
uint32 checksum_version, uint32 backup_version);
/* parsexlog.c */
extern void extractPageMap(const char *archivedir,
TimeLineID tli, uint32 seg_size,
XLogRecPtr startpoint, XLogRecPtr endpoint);
extern bool extractPageMap(const char *archivedir, uint32 wal_seg_size,
XLogRecPtr startpoint, TimeLineID start_tli,
XLogRecPtr endpoint, TimeLineID end_tli,
parray *tli_list);
extern void validate_wal(pgBackup *backup, const char *archivedir,
time_t target_time, TransactionId target_xid,
XLogRecPtr target_lsn, TimeLineID tli,
src/restore.c
+2 -2
View File
@@ -403,7 +403,7 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
*/
validate_wal(dest_backup, arclog_path, rt->target_time,
rt->target_xid, rt->target_lsn,
base_full_backup->tli, instance_config.xlog_seg_size);
dest_backup->tli, instance_config.xlog_seg_size);
}
/* Orphanize every OK descendant of corrupted backup */
else
@@ -1326,7 +1326,7 @@ satisfy_timeline(const parray *timelines, const pgBackup *backup)
timeline = (TimeLineHistoryEntry *) parray_get(timelines, i);
if (backup->tli == timeline->tli &&
(XLogRecPtrIsInvalid(timeline->end) ||
backup->stop_lsn < timeline->end))
backup->stop_lsn <= timeline->end))
return true;
}
return false;
tests/archive.py
+1 -1
View File
@@ -1761,7 +1761,7 @@ class ArchiveTest(ProbackupTest, unittest.TestCase):
tli13['closest-backup-id'])
self.assertEqual(
'0000000D000000000000001B',
'0000000D000000000000001C',
tli13['max-segno'])
# Clean after yourself
tests/backup.py
+30 -23
View File
@@ -228,10 +228,9 @@ class BackupTest(ProbackupTest, unittest.TestCase):
"without valid full backup.\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
"ERROR: Valid backup on current timeline 1 is not found. "
"Create new FULL backup before an incremental one.",
e.message,
self.assertTrue(
"WARNING: Valid backup on current timeline 1 is not found" in e.message and
"ERROR: Create new full backup before an incremental one" in e.message,
"\n Unexpected Error Message: {0}\n CMD: {1}".format(
repr(e.message), self.cmd))
@@ -2294,10 +2293,9 @@ class BackupTest(ProbackupTest, unittest.TestCase):
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'ERROR: Valid backup on current timeline 1 is not found. '
'Create new FULL backup before an incremental one.',
e.message,
self.assertTrue(
'WARNING: Valid backup on current timeline 1 is not found' in e.message and
'ERROR: Create new full backup before an incremental one' in e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
@@ -2324,10 +2322,13 @@ class BackupTest(ProbackupTest, unittest.TestCase):
initdb_params=['--data-checksums'],
pg_options={
'archive_timeout': '30s',
'checkpoint_timeout': '1h'})
'archive_mode': 'always',
'checkpoint_timeout': '60s',
'wal_level': 'logical'})
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_config(backup_dir, 'node', options=['--archive-timeout=60s'])
self.set_archiving(backup_dir, 'node', node)
node.slow_start()
@@ -2447,12 +2448,15 @@ class BackupTest(ProbackupTest, unittest.TestCase):
self.restore_node(backup_dir, 'node', replica)
self.set_replica(node, replica)
self.add_instance(backup_dir, 'replica', replica)
self.set_config(
backup_dir, 'replica',
options=['--archive-timeout=120s', '--log-level-console=LOG'])
self.set_archiving(backup_dir, 'replica', replica, replica=True)
self.set_auto_conf(replica, {'hot_standby': 'on'})
# freeze bgwriter to get rid of RUNNING XACTS records
bgwriter_pid = node.auxiliary_pids[ProcessType.BackgroundWriter][0]
gdb_checkpointer = self.gdb_attach(bgwriter_pid)
# bgwriter_pid = node.auxiliary_pids[ProcessType.BackgroundWriter][0]
# gdb_checkpointer = self.gdb_attach(bgwriter_pid)
copy_tree(
os.path.join(backup_dir, 'wal', 'node'),
@@ -2460,21 +2464,22 @@ class BackupTest(ProbackupTest, unittest.TestCase):
replica.slow_start(replica=True)
self.switch_wal_segment(node)
self.switch_wal_segment(node)
# self.switch_wal_segment(node)
# self.switch_wal_segment(node)
# FULL backup from replica
self.backup_node(
backup_dir, 'replica', replica,
datname='backupdb', options=['--stream', '-U', 'backup', '--archive-timeout=30s'])
datname='backupdb', options=['-U', 'backup'])
# stream full backup from replica
self.backup_node(
backup_dir, 'replica', replica,
datname='backupdb', options=['--stream', '-U', 'backup'])
# self.switch_wal_segment(node)
self.backup_node(
backup_dir, 'replica', replica, datname='backupdb',
options=['-U', 'backup', '--archive-timeout=300s'])
# PAGE backup from replica
self.switch_wal_segment(node)
self.backup_node(
backup_dir, 'replica', replica, backup_type='page',
datname='backupdb', options=['-U', 'backup', '--archive-timeout=30s'])
@@ -2484,20 +2489,22 @@ class BackupTest(ProbackupTest, unittest.TestCase):
datname='backupdb', options=['--stream', '-U', 'backup'])
# DELTA backup from replica
self.switch_wal_segment(node)
self.backup_node(
backup_dir, 'replica', replica, backup_type='delta',
datname='backupdb', options=['-U', 'backup', '--archive-timeout=30s'])
datname='backupdb', options=['-U', 'backup'])
self.backup_node(
backup_dir, 'replica', replica, backup_type='delta',
datname='backupdb', options=['--stream', '-U', 'backup'])
# PTRACK backup from replica
if self.ptrack:
self.switch_wal_segment(node)
self.backup_node(
backup_dir, 'replica', replica, backup_type='delta',
datname='backupdb', options=['-U', 'backup', '--archive-timeout=30s'])
backup_dir, 'replica', replica, backup_type='ptrack',
datname='backupdb', options=['-U', 'backup'])
self.backup_node(
backup_dir, 'replica', replica, backup_type='delta',
backup_dir, 'replica', replica, backup_type='ptrack',
datname='backupdb', options=['--stream', '-U', 'backup'])
# Clean after yourself
tests/page.py
+83 -6
View File
@@ -874,7 +874,6 @@ class PageTest(ProbackupTest, unittest.TestCase):
'INFO: Wait for WAL segment' in e.message and
'to be archived' in e.message and
'Could not read WAL record at' in e.message and
'incorrect resource manager data checksum in record at' in e.message and
'Possible WAL corruption. Error has occured during reading WAL segment' in e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
@@ -899,7 +898,6 @@ class PageTest(ProbackupTest, unittest.TestCase):
'INFO: Wait for WAL segment' in e.message and
'to be archived' in e.message and
'Could not read WAL record at' in e.message and
'incorrect resource manager data checksum in record at' in e.message and
'Possible WAL corruption. Error has occured during reading WAL segment "{0}"'.format(
file) in e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
@@ -942,8 +940,10 @@ class PageTest(ProbackupTest, unittest.TestCase):
self.set_archiving(backup_dir, 'alien_node', alien_node)
alien_node.slow_start()
self.backup_node(backup_dir, 'node', node)
self.backup_node(backup_dir, 'alien_node', alien_node)
self.backup_node(
backup_dir, 'node', node, options=['--stream'])
self.backup_node(
backup_dir, 'alien_node', alien_node, options=['--stream'])
# make some wals
node.safe_psql(
@@ -996,8 +996,6 @@ class PageTest(ProbackupTest, unittest.TestCase):
'INFO: Wait for WAL segment' in e.message and
'to be archived' in e.message and
'Could not read WAL record at' in e.message and
'WAL file is from different database system: WAL file database system identifier is' in e.message and
'pg_control database system identifier is' in e.message and
'Possible WAL corruption. Error has occured during reading WAL segment' in e.message,
'\n Unexpected Error Message: {0}\n CMD: {1}'.format(
repr(e.message), self.cmd))
@@ -1181,6 +1179,85 @@ class PageTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
# @unittest.expectedFailure
def test_multi_timeline_page(self):
"""
Check that backup in PAGE mode choose
parent backup correctly:
t12 /---P-->
...
t3 /---->
t2 /---->
t1 -F-----D->
P must have F as parent
"""
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()
node.pgbench_init(scale=50)
full_id = self.backup_node(backup_dir, 'node', node)
pgbench = node.pgbench(options=['-T', '20', '-c', '1', '--no-vacuum'])
pgbench.wait()
self.backup_node(backup_dir, 'node', node, backup_type='delta')
node.cleanup()
self.restore_node(
backup_dir, 'node', node, backup_id=full_id,
options=[
'--recovery-target=immediate',
'--recovery-target-action=promote'])
node.slow_start()
pgbench = node.pgbench(options=['-T', '20', '-c', '1', '--no-vacuum'])
pgbench.wait()
# create timelines
for i in range(2, 12):
node.cleanup()
self.restore_node(
backup_dir, 'node', node, backup_id=full_id,
options=['--recovery-target-timeline={0}'.format(i)])
node.slow_start()
pgbench = node.pgbench(options=['-T', '3', '-c', '1', '--no-vacuum'])
pgbench.wait()
page_id = self.backup_node(
backup_dir, 'node', node, backup_type='page',
options=['--log-level-file=VERBOSE'])
pgdata = self.pgdata_content(node.data_dir)
node.cleanup()
self.restore_node(backup_dir, 'node', node)
pgdata_restored = self.pgdata_content(node.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
show = self.show_archive(backup_dir)
timelines = show[0]['timelines']
# self.assertEqual()
self.assertEqual(
self.show_pb(backup_dir, 'node', page_id)['parent-backup-id'],
full_id)
# Clean after yourself
self.del_test_dir(module_name, fname)
@unittest.skip("skip")
# @unittest.expectedFailure
def test_page_pg_resetxlog(self):
tests/replica.py
+564 -57
View File
@@ -571,30 +571,25 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
'Skipped because backup from replica is not supported in PG 9.5')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'master', master)
self.set_archiving(backup_dir, 'master', master)
self.add_instance(backup_dir, 'node', master)
self.set_archiving(backup_dir, 'node', master)
master.slow_start()
# freeze bgwriter to get rid of RUNNING XACTS records
bgwriter_pid = master.auxiliary_pids[ProcessType.BackgroundWriter][0]
gdb_checkpointer = self.gdb_attach(bgwriter_pid)
self.backup_node(backup_dir, 'master', master)
self.backup_node(backup_dir, 'node', master)
# Create replica
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(backup_dir, 'master', replica)
self.restore_node(backup_dir, 'node', replica)
# Settings for Replica
self.add_instance(backup_dir, 'replica', replica)
self.set_replica(master, replica, synchronous=True)
self.set_archiving(backup_dir, 'replica', replica, replica=True)
copy_tree(
os.path.join(backup_dir, 'wal', 'master'),
os.path.join(backup_dir, 'wal', 'replica'))
self.set_archiving(backup_dir, 'node', replica, replica=True)
replica.slow_start(replica=True)
@@ -602,7 +597,7 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
self.switch_wal_segment(master)
output = self.backup_node(
backup_dir, 'replica', replica,
backup_dir, 'node', replica, replica.data_dir,
options=[
'--archive-timeout=30',
'--log-level-console=LOG',
@@ -611,24 +606,24 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
return_id=False)
self.assertIn(
'LOG: Null offset in stop_backup_lsn value 0/3000000',
'LOG: Null offset in stop_backup_lsn value 0/4000000',
output)
self.assertIn(
'WARNING: WAL segment 000000010000000000000003 could not be streamed in 30 seconds',
'WARNING: WAL segment 000000010000000000000004 could not be streamed in 30 seconds',
output)
self.assertIn(
'WARNING: Failed to get next WAL record after 0/3000000, looking for previous WAL record',
'WARNING: Failed to get next WAL record after 0/4000000, looking for previous WAL record',
output)
self.assertIn(
'LOG: Looking for LSN 0/3000000 in segment: 000000010000000000000002',
'LOG: Looking for LSN 0/4000000 in segment: 000000010000000000000003',
output)
self.assertIn(
'has endpoint 0/3000000 which is '
'equal or greater than requested LSN 0/3000000',
'has endpoint 0/4000000 which is '
'equal or greater than requested LSN 0/4000000',
output)
self.assertIn(
@@ -719,19 +714,19 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
log_content = f.read()
self.assertIn(
'LOG: Null offset in stop_backup_lsn value 0/3000000',
'LOG: Null offset in stop_backup_lsn value 0/4000000',
log_content)
self.assertIn(
'LOG: Looking for segment: 000000010000000000000003',
'LOG: Looking for segment: 000000010000000000000004',
log_content)
self.assertIn(
'LOG: First record in WAL segment "000000010000000000000003": 0/3000028',
'LOG: First record in WAL segment "000000010000000000000004": 0/4000028',
log_content)
self.assertIn(
'LOG: current.stop_lsn: 0/3000028',
'LOG: current.stop_lsn: 0/4000028',
log_content)
# Clean after yourself
@@ -757,31 +752,26 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
'Skipped because backup from replica is not supported in PG 9.5')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'master', master)
self.set_archiving(backup_dir, 'master', master)
self.add_instance(backup_dir, 'node', master)
self.set_archiving(backup_dir, 'node', master)
master.slow_start()
self.backup_node(backup_dir, 'master', master)
self.backup_node(backup_dir, 'node', master)
# Create replica
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(backup_dir, 'master', replica)
self.restore_node(backup_dir, 'node', replica)
# Settings for Replica
self.add_instance(backup_dir, 'replica', replica)
self.set_replica(master, replica, synchronous=True)
self.set_archiving(backup_dir, 'replica', replica, replica=True)
self.set_archiving(backup_dir, 'node', replica, replica=True)
# freeze bgwriter to get rid of RUNNING XACTS records
bgwriter_pid = master.auxiliary_pids[ProcessType.BackgroundWriter][0]
gdb_checkpointer = self.gdb_attach(bgwriter_pid)
copy_tree(
os.path.join(backup_dir, 'wal', 'master'),
os.path.join(backup_dir, 'wal', 'replica'))
replica.slow_start(replica=True)
self.switch_wal_segment(master)
@@ -789,7 +779,7 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
# take backup from replica
output = self.backup_node(
backup_dir, 'replica', replica,
backup_dir, 'node', replica, replica.data_dir,
options=[
'--archive-timeout=30',
'--log-level-console=LOG',
@@ -797,24 +787,24 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
return_id=False)
self.assertIn(
'LOG: Null offset in stop_backup_lsn value 0/3000000',
'LOG: Null offset in stop_backup_lsn value 0/4000000',
output)
self.assertIn(
'WARNING: WAL segment 000000010000000000000003 could not be archived in 30 seconds',
'WARNING: WAL segment 000000010000000000000004 could not be archived in 30 seconds',
output)
self.assertIn(
'WARNING: Failed to get next WAL record after 0/3000000, looking for previous WAL record',
'WARNING: Failed to get next WAL record after 0/4000000, looking for previous WAL record',
output)
self.assertIn(
'LOG: Looking for LSN 0/3000000 in segment: 000000010000000000000002',
'LOG: Looking for LSN 0/4000000 in segment: 000000010000000000000003',
output)
self.assertIn(
'has endpoint 0/3000000 which is '
'equal or greater than requested LSN 0/3000000',
'has endpoint 0/4000000 which is '
'equal or greater than requested LSN 0/4000000',
output)
self.assertIn(
@@ -846,44 +836,39 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
'Skipped because backup from replica is not supported in PG 9.5')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'master', master)
self.set_archiving(backup_dir, 'master', master)
self.add_instance(backup_dir, 'node', master)
self.set_archiving(backup_dir, 'node', master)
master.slow_start()
self.backup_node(backup_dir, 'master', master)
self.backup_node(backup_dir, 'node', master)
# Create replica
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(backup_dir, 'master', replica)
self.restore_node(backup_dir, 'node', replica)
# Settings for Replica
self.add_instance(backup_dir, 'replica', replica)
self.set_replica(master, replica, synchronous=True)
self.set_archiving(backup_dir, 'replica', replica, replica=True)
copy_tree(
os.path.join(backup_dir, 'wal', 'master'),
os.path.join(backup_dir, 'wal', 'replica'))
self.set_archiving(backup_dir, 'node', replica, replica=True)
replica.slow_start(replica=True)
# take backup from replica
self.backup_node(
backup_dir, 'replica', replica,
backup_dir, 'node', replica, replica.data_dir,
options=[
'--archive-timeout=30',
'--log-level-console=verbose',
'--log-level-console=LOG',
'--no-validate'],
return_id=False)
try:
self.backup_node(
backup_dir, 'replica', replica,
backup_dir, 'node', replica, replica.data_dir,
options=[
'--archive-timeout=30',
'--log-level-console=verbose',
'--log-level-console=LOG',
'--no-validate'])
# we should die here because exception is what we expect to happen
self.assertEqual(
@@ -893,19 +878,19 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
'LOG: Looking for LSN 0/3000060 in segment: 000000010000000000000003',
'LOG: Looking for LSN 0/4000060 in segment: 000000010000000000000004',
e.message,
"\n Unexpected Error Message: {0}\n CMD: {1}".format(
repr(e.message), self.cmd))
self.assertIn(
'INFO: Wait for LSN 0/3000060 in archived WAL segment',
'INFO: Wait for LSN 0/4000060 in archived WAL segment',
e.message,
"\n Unexpected Error Message: {0}\n CMD: {1}".format(
repr(e.message), self.cmd))
self.assertIn(
'ERROR: WAL segment 000000010000000000000003 could not be archived in 30 seconds',
'ERROR: WAL segment 000000010000000000000004 could not be archived in 30 seconds',
e.message,
"\n Unexpected Error Message: {0}\n CMD: {1}".format(
repr(e.message), self.cmd))
@@ -1016,7 +1001,7 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
@unittest.skip("skip")
def test_replica_promote_1(self):
"""
"""
@@ -1037,7 +1022,7 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'master', master)
# set replica True, so archive_mode 'always' is used.
# set replica True, so archive_mode 'always' is used.
self.set_archiving(backup_dir, 'master', master, replica=True)
master.slow_start()
@@ -1091,6 +1076,528 @@ class ReplicaTest(ProbackupTest, unittest.TestCase):
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_replica_promote_2(self):
"""
"""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
master = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'master'),
set_replication=True,
initdb_params=['--data-checksums'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'master', master)
# set replica True, so archive_mode 'always' is used.
self.set_archiving(
backup_dir, 'master', master, replica=True)
master.slow_start()
self.backup_node(backup_dir, 'master', master)
# Create replica
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(backup_dir, 'master', replica)
# Settings for Replica
self.set_replica(master, replica)
self.set_auto_conf(replica, {'port': replica.port})
replica.slow_start(replica=True)
master.safe_psql(
'postgres',
'CREATE TABLE t1 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,1) i')
self.wait_until_replica_catch_with_master(master, replica)
replica.promote()
replica.safe_psql(
'postgres',
'CHECKPOINT')
# replica.safe_psql(
# 'postgres',
# 'create table t2()')
#
# replica.safe_psql(
# 'postgres',
# 'CHECKPOINT')
self.backup_node(
backup_dir, 'master', replica, data_dir=replica.data_dir,
backup_type='page')
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_replica_promote_3(self):
"""
"""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
master = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'master'),
set_replication=True,
initdb_params=['--data-checksums'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'master', master)
master.slow_start()
self.backup_node(backup_dir, 'master', master, options=['--stream'])
# Create replica
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(backup_dir, 'master', replica)
# Settings for Replica
self.set_replica(master, replica)
self.set_auto_conf(replica, {'port': replica.port})
replica.slow_start(replica=True)
master.safe_psql(
'postgres',
'CREATE TABLE t1 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(master, replica)
self.add_instance(backup_dir, 'replica', replica)
replica.safe_psql(
'postgres',
'CHECKPOINT')
full_id = self.backup_node(
backup_dir, 'replica',
replica, options=['--stream'])
master.safe_psql(
'postgres',
'CREATE TABLE t2 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(master, replica)
replica.safe_psql(
'postgres',
'CHECKPOINT')
self.backup_node(
backup_dir, 'replica', replica,
backup_type='delta', options=['--stream'])
replica.promote()
replica.safe_psql(
'postgres',
'CHECKPOINT')
# failing, because without archving, it is impossible to
# take multi-timeline backup.
try:
self.backup_node(
backup_dir, 'replica', replica,
backup_type='delta', options=['--stream'])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of timeline switch "
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertTrue(
'WARNING: Cannot find valid backup on previous timelines, '
'WAL archive is not available' in e.message and
'ERROR: Create new full backup before an incremental one' 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_replica_promote_archive_delta(self):
"""
t3 /---D3-->
t2 /------->
t1 --F---D1--D2--
"""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
node1 = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node1'),
set_replication=True,
initdb_params=['--data-checksums'],
pg_options={
'checkpoint_timeout': '30s',
'archive_timeout': '30s',
'autovacuum': 'off'})
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node1)
self.set_config(
backup_dir, 'node', options=['--archive-timeout=60s'])
self.set_archiving(backup_dir, 'node', node1)
node1.slow_start()
self.backup_node(backup_dir, 'node', node1, options=['--stream'])
# Create replica
node2 = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node2'))
node2.cleanup()
self.restore_node(backup_dir, 'node', node2, node2.data_dir)
# Settings for Replica
self.set_replica(node1, node2)
self.set_auto_conf(node2, {'port': node2.port})
self.set_archiving(backup_dir, 'node', node2, replica=True)
node2.slow_start(replica=True)
node1.safe_psql(
'postgres',
'CREATE TABLE t1 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(node1, node2)
node1.safe_psql(
'postgres',
'CREATE TABLE t2 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(node1, node2)
# delta backup on replica on timeline 1
delta1_id = self.backup_node(
backup_dir, 'node', node2, node2.data_dir,
'delta', options=['--stream'])
# delta backup on replica on timeline 1
delta2_id = self.backup_node(
backup_dir, 'node', node2, node2.data_dir, 'delta')
self.change_backup_status(
backup_dir, 'node', delta2_id, 'ERROR')
# node2 is now master
node2.promote()
node2.safe_psql('postgres', 'CHECKPOINT')
node2.safe_psql(
'postgres',
'CREATE TABLE t3 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
# node1 is now replica
node1.cleanup()
# kludge "backup_id=delta1_id"
self.restore_node(
backup_dir, 'node', node1, node1.data_dir,
backup_id=delta1_id,
options=[
'--recovery-target-timeline=2',
'--recovery-target=latest'])
# Settings for Replica
self.set_replica(node2, node1)
self.set_auto_conf(node1, {'port': node1.port})
self.set_archiving(backup_dir, 'node', node1, replica=True)
node1.slow_start(replica=True)
node2.safe_psql(
'postgres',
'CREATE TABLE t4 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,30) i')
self.wait_until_replica_catch_with_master(node2, node1)
# node1 is back to be a master
node1.promote()
node1.safe_psql('postgres', 'CHECKPOINT')
# delta backup on timeline 3
self.backup_node(
backup_dir, 'node', node1, node1.data_dir, 'delta',
options=['--archive-timeout=60'])
pgdata = self.pgdata_content(node1.data_dir)
node1.cleanup()
self.restore_node(backup_dir, 'node', node1, node1.data_dir)
pgdata_restored = self.pgdata_content(node1.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_replica_promote_archive_page(self):
"""
t3 /---P3-->
t2 /------->
t1 --F---P1--P2--
"""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
node1 = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node1'),
set_replication=True,
initdb_params=['--data-checksums'],
pg_options={
'checkpoint_timeout': '30s',
'archive_timeout': '30s',
'autovacuum': 'off'})
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node1)
self.set_archiving(backup_dir, 'node', node1)
self.set_config(
backup_dir, 'node', options=['--archive-timeout=60s'])
node1.slow_start()
self.backup_node(backup_dir, 'node', node1, options=['--stream'])
# Create replica
node2 = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node2'))
node2.cleanup()
self.restore_node(backup_dir, 'node', node2, node2.data_dir)
# Settings for Replica
self.set_replica(node1, node2)
self.set_auto_conf(node2, {'port': node2.port})
self.set_archiving(backup_dir, 'node', node2, replica=True)
node2.slow_start(replica=True)
node1.safe_psql(
'postgres',
'CREATE TABLE t1 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(node1, node2)
node1.safe_psql(
'postgres',
'CREATE TABLE t2 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(node1, node2)
# page backup on replica on timeline 1
page1_id = self.backup_node(
backup_dir, 'node', node2, node2.data_dir,
'page', options=['--stream'])
# page backup on replica on timeline 1
page2_id = self.backup_node(
backup_dir, 'node', node2, node2.data_dir, 'page')
self.change_backup_status(
backup_dir, 'node', page2_id, 'ERROR')
# node2 is now master
node2.promote()
node2.safe_psql('postgres', 'CHECKPOINT')
node2.safe_psql(
'postgres',
'CREATE TABLE t3 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
# node1 is now replica
node1.cleanup()
# kludge "backup_id=page1_id"
self.restore_node(
backup_dir, 'node', node1, node1.data_dir,
backup_id=page1_id,
options=[
'--recovery-target-timeline=2',
'--recovery-target=latest'])
# Settings for Replica
self.set_replica(node2, node1)
self.set_auto_conf(node1, {'port': node1.port})
self.set_archiving(backup_dir, 'node', node1, replica=True)
node1.slow_start(replica=True)
node2.safe_psql(
'postgres',
'CREATE TABLE t4 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,30) i')
self.wait_until_replica_catch_with_master(node2, node1)
# node1 is back to be a master
node1.promote()
node1.safe_psql('postgres', 'CHECKPOINT')
# delta3_id = self.backup_node(
# backup_dir, 'node', node2, node2.data_dir, 'delta')
# page backup on timeline 3
page3_id = self.backup_node(
backup_dir, 'node', node1, node1.data_dir, 'page',
options=['--archive-timeout=60'])
pgdata = self.pgdata_content(node1.data_dir)
node1.cleanup()
self.restore_node(backup_dir, 'node', node1, node1.data_dir)
pgdata_restored = self.pgdata_content(node1.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
# Clean after yourself
self.del_test_dir(module_name, fname)
# @unittest.skip("skip")
def test_parent_choosing(self):
"""
"""
fname = self.id().split('.')[3]
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
master = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'master'),
set_replication=True,
initdb_params=['--data-checksums'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'master', master)
master.slow_start()
self.backup_node(backup_dir, 'master', master, options=['--stream'])
# Create replica
replica = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'replica'))
replica.cleanup()
self.restore_node(backup_dir, 'master', replica)
# Settings for Replica
self.set_replica(master, replica)
self.set_auto_conf(replica, {'port': replica.port})
replica.slow_start(replica=True)
master.safe_psql(
'postgres',
'CREATE TABLE t1 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(master, replica)
self.add_instance(backup_dir, 'replica', replica)
full_id = self.backup_node(
backup_dir, 'replica',
replica, options=['--stream'])
master.safe_psql(
'postgres',
'CREATE TABLE t2 AS '
'SELECT i, repeat(md5(i::text),5006056) AS fat_attr '
'FROM generate_series(0,20) i')
self.wait_until_replica_catch_with_master(master, replica)
self.backup_node(
backup_dir, 'replica', replica,
backup_type='delta', options=['--stream'])
replica.promote()
replica.safe_psql('postgres', 'CHECKPOINT')
# failing, because without archving, it is impossible to
# take multi-timeline backup.
try:
self.backup_node(
backup_dir, 'replica', replica,
backup_type='delta', options=['--stream'])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because of timeline switch "
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertTrue(
'WARNING: Cannot find valid backup on previous timelines, '
'WAL archive is not available' in e.message and
'ERROR: Create new full backup before an incremental one' 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_instance_from_the_past(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'])
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
node.slow_start()
full_id = self.backup_node(backup_dir, 'node', node, options=['--stream'])
node.pgbench_init(scale=10)
self.backup_node(backup_dir, 'node', node, options=['--stream'])
node.cleanup()
self.restore_node(backup_dir, 'node', node, backup_id=full_id)
node.slow_start()
try:
self.backup_node(
backup_dir, 'node', node,
backup_type='delta', options=['--stream'])
# we should die here because exception is what we expect to happen
self.assertEqual(
1, 0,
"Expecting Error because instance is from the past "
"\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertTrue(
'ERROR: Current START LSN' in e.message and
'is lower than START LSN' in e.message and
'It may indicate that we are trying to backup '
'PostgreSQL instance from the past' 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)
# TODO:
# null offset STOP LSN and latest record in previous segment is conrecord (manual only)
tests/retention.py
+11 -12
View File
@@ -1712,10 +1712,9 @@ class RetentionTest(ProbackupTest, unittest.TestCase):
"without valid full backup.\n Output: {0} \n CMD: {1}".format(
repr(self.output), self.cmd))
except ProbackupException as e:
self.assertIn(
"ERROR: Valid backup on current timeline 1 is not found. "
"Create new FULL backup before an incremental one.",
e.message,
self.assertTrue(
"WARNING: Valid backup on current timeline 1 is not found" in e.message and
"ERROR: Create new full backup before an incremental one" in e.message,
"\n Unexpected Error Message: {0}\n CMD: {1}".format(
repr(e.message), self.cmd))
@@ -2675,7 +2674,7 @@ class RetentionTest(ProbackupTest, unittest.TestCase):
self.assertIn(
'LOG: Archive backup {0} to stay consistent protect from '
'purge WAL interval between 000000010000000000000004 '
'and 000000010000000000000004 on timeline 1'.format(B1), output)
'and 000000010000000000000005 on timeline 1'.format(B1), output)
start_lsn_B4 = self.show_pb(backup_dir, 'node', B4)['start-lsn']
self.assertIn(
@@ -2684,13 +2683,13 @@ class RetentionTest(ProbackupTest, unittest.TestCase):
self.assertIn(
'LOG: Timeline 3 to stay reachable from timeline 1 protect '
'from purge WAL interval between 000000020000000000000005 and '
'000000020000000000000008 on timeline 2', output)
'from purge WAL interval between 000000020000000000000006 and '
'000000020000000000000009 on timeline 2', output)
self.assertIn(
'LOG: Timeline 3 to stay reachable from timeline 1 protect '
'from purge WAL interval between 000000010000000000000004 and '
'000000010000000000000005 on timeline 1', output)
'000000010000000000000006 on timeline 1', output)
show_tli1_before = self.show_archive(backup_dir, 'node', tli=1)
show_tli2_before = self.show_archive(backup_dir, 'node', tli=2)
@@ -2745,19 +2744,19 @@ class RetentionTest(ProbackupTest, unittest.TestCase):
self.assertEqual(
show_tli1_after['lost-segments'][0]['begin-segno'],
'000000010000000000000006')
'000000010000000000000007')
self.assertEqual(
show_tli1_after['lost-segments'][0]['end-segno'],
'000000010000000000000009')
'00000001000000000000000A')
self.assertEqual(
show_tli2_after['lost-segments'][0]['begin-segno'],
'000000020000000000000009')
'00000002000000000000000A')
self.assertEqual(
show_tli2_after['lost-segments'][0]['end-segno'],
'000000020000000000000009')
'00000002000000000000000A')
self.validate_pb(backup_dir, 'node')