db/pg_probackup
db/pg_probackup
1
0
Fork 0
mirror of https://github.com/postgrespro/pg_probackup.git synced 2024-11-24 08:52:38 +02:00
Code Issues Releases Activity
0e11c87396
pg_probackup/backup.c
Anastasia d40e299c3c fix dest backup search
2017-05-02 17:28:08 +03:00

1709 lines
44 KiB
C
Raw Blame History

/*-------------------------------------------------------------------------
*
* backup.c: backup DB cluster, archived WAL
*
* Portions Copyright (c) 2009-2013, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
* Portions Copyright (c) 2015-2017, Postgres Professional
*
*-------------------------------------------------------------------------
*/
#include "pg_probackup.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <unistd.h>
#include <dirent.h>
#include <time.h>
#include <pthread.h>
#include "libpq/pqsignal.h"
#include "storage/bufpage.h"
#include "datapagemap.h"
#include "streamutil.h"
#include "receivelog.h"
static int standby_message_timeout = 10 * 1000; /* 10 sec = default */
static XLogRecPtr stop_backup_lsn = InvalidXLogRecPtr;
const char *progname = "pg_probackup";
/* list of files contained in backup */
static parray *backup_files_list = NULL;
static pthread_mutex_t check_stream_mut = PTHREAD_MUTEX_INITIALIZER;
static int is_ptrack_enable = false;
/* Backup connection */
static PGconn *backup_conn = NULL;
typedef struct
{
const char *from_root;
const char *to_root;
parray *backup_files_list;
parray *prev_backup_filelist;
XLogRecPtr prev_backup_start_lsn;
} backup_files_args;
/*
* Backup routines
*/
static void backup_cleanup(bool fatal, void *userdata);
static void backup_disconnect(bool fatal, void *userdata);
static void backup_files(void *arg);
static void do_backup_database(parray *backup_list);
static void pg_start_backup(const char *label, bool smooth, pgBackup *backup);
static void pg_switch_wal(void);
static void pg_stop_backup(pgBackup *backup);
static void add_pgdata_files(parray *files, const char *root);
static void write_backup_file_list(parray *files, const char *root);
static void wait_archive_lsn(XLogRecPtr lsn, bool prev_segno);
static void make_pagemap_from_ptrack(parray *files);
static void StreamLog(void *arg);
/* Ptrack functions */
static void pg_ptrack_clear(void);
static bool pg_ptrack_support(void);
static bool pg_ptrack_enable(void);
static bool pg_is_in_recovery(void);
static char *pg_ptrack_get_and_clear(Oid tablespace_oid,
Oid db_oid,
Oid rel_oid,
size_t *result_size);
/* Check functions */
static void check_server_version(void);
static void check_system_identifiers(void);
static void confirm_block_size(const char *name, int blcksz);
#define disconnect_and_exit(code) \
{ \
if (conn != NULL) PQfinish(conn); \
exit(code); \
}
/*
* Take a backup of database.
*/
static void
do_backup_database(parray *backup_list)
{
size_t i;
char database_path[MAXPGPATH];
char dst_backup_path[MAXPGPATH];
char label[1024];
XLogRecPtr prev_backup_start_lsn = InvalidXLogRecPtr;
pthread_t backup_threads[num_threads];
pthread_t stream_thread;
backup_files_args *backup_threads_args[num_threads];
pgBackup *prev_backup = NULL;
char prev_backup_filelist_path[MAXPGPATH];
parray *prev_backup_filelist = NULL;
elog(LOG, "Database backup start");
/* Initialize size summary */
current.data_bytes = 0;
/* Obtain current timeline from PGDATA control file */
current.tli = get_current_timeline(false);
/*
* In incremental backup mode ensure that already-validated
* backup on current timeline exists.
*/
if (current.backup_mode == BACKUP_MODE_DIFF_PAGE ||
current.backup_mode == BACKUP_MODE_DIFF_PTRACK)
{
prev_backup = catalog_get_last_data_backup(backup_list, current.tli);
if (prev_backup == NULL)
elog(ERROR, "Valid backup on current timeline is not found. "
"Create new FULL backup before an incremental one.");
}
/* Clear ptrack files for FULL and PAGE backup */
if (current.backup_mode != BACKUP_MODE_DIFF_PTRACK && is_ptrack_enable)
pg_ptrack_clear();
/* notify start of backup to PostgreSQL server */
time2iso(label, lengthof(label), current.start_time);
strncat(label, " with pg_probackup", lengthof(label));
pg_start_backup(label, smooth_checkpoint, &current);
pgBackupGetPath(&current, database_path, lengthof(database_path),
DATABASE_DIR);
/* start stream replication */
if (stream_wal)
{
join_path_components(dst_backup_path, database_path, PG_XLOG_DIR);
dir_create_dir(dst_backup_path, DIR_PERMISSION);
pthread_mutex_lock(&check_stream_mut);
pthread_create(&stream_thread, NULL, (void *(*)(void *)) StreamLog, dst_backup_path);
pthread_mutex_lock(&check_stream_mut);
if (conn == NULL)
elog(ERROR, "Cannot continue backup because stream connect has failed.");
pthread_mutex_unlock(&check_stream_mut);
}
/*
* If backup_label does not exist in $PGDATA, stop taking backup.
* NOTE. We can check it only on master, though.
*/
if(!from_replica)
{
char label_path[MAXPGPATH];
join_path_components(label_path, pgdata, PG_BACKUP_LABEL_FILE);
/* Leave if no backup file */
if (!fileExists(label_path))
{
elog(LOG, "%s does not exist, stopping backup", PG_BACKUP_LABEL_FILE);
pg_stop_backup(NULL);
elog(ERROR, "%s does not exist in PGDATA", PG_BACKUP_LABEL_FILE);
}
}
/*
* To take incremental backup get the filelist of the last completed database
*/
if (current.backup_mode == BACKUP_MODE_DIFF_PAGE ||
current.backup_mode == BACKUP_MODE_DIFF_PTRACK)
{
Assert(prev_backup);
pgBackupGetPath(prev_backup, prev_backup_filelist_path, lengthof(prev_backup_filelist_path),
DATABASE_FILE_LIST);
prev_backup_filelist = dir_read_file_list(pgdata, prev_backup_filelist_path);
/* If lsn is not NULL, only pages with higher lsn will be copied. */
prev_backup_start_lsn = prev_backup->start_lsn;
current.parent_backup = prev_backup->start_time;
pgBackupWriteBackupControlFile(&current);
}
/* initialize backup list */
backup_files_list = parray_new();
/* list files with the logical path. omit $PGDATA */
add_pgdata_files(backup_files_list, pgdata);
if (current.backup_mode != BACKUP_MODE_FULL)
{
elog(LOG, "current_tli:%X", current.tli);
elog(LOG, "prev_backup->start_lsn: %X/%X",
(uint32) (prev_backup->start_lsn >> 32), (uint32) (prev_backup->start_lsn));
elog(LOG, "current.start_lsn: %X/%X",
(uint32) (current.start_lsn >> 32), (uint32) (current.start_lsn));
}
/*
* Build page mapping in incremental mode.
*/
if (current.backup_mode == BACKUP_MODE_DIFF_PAGE)
{
/*
* Switch to a new WAL segment. It is necessary to get archived WAL
* segment, which includes start LSN of current backup.
*
* Do not switch for standby node and if backup is stream.
*/
if (!from_replica && !stream_wal)
pg_switch_wal();
if (!stream_wal)
wait_archive_lsn(current.start_lsn, false);
/*
* Build the page map. Obtain information about changed pages
* reading WAL segments present in archives up to the point
* where this backup has started.
*/
extractPageMap(arclog_path, prev_backup->start_lsn, current.tli,
current.start_lsn,
/*
* For backup from master wait for previous segment.
* For backup from replica wait for current segment.
*/
!from_replica);
}
else if (current.backup_mode == BACKUP_MODE_DIFF_PTRACK)
{
XLogRecPtr ptrack_lsn = get_last_ptrack_lsn();
if (ptrack_lsn > prev_backup->stop_lsn)
{
elog(ERROR, "LSN from ptrack_control %lx differs from LSN of previous ptrack backup %lx.\n"
"Create new full backup before an incremental one.",
ptrack_lsn, prev_backup->start_lsn);
}
parray_qsort(backup_files_list, pgFileComparePathDesc);
make_pagemap_from_ptrack(backup_files_list);
}
/*
* Sort pathname ascending. It is necessary to create intermediate
* directories sequentially.
*
* For example:
* 1 - create 'base'
* 2 - create 'base/1'
*/
parray_qsort(backup_files_list, pgFileComparePath);
/*
* Make directories before backup
* and setup threads at the same time
*/
for (i = 0; i < parray_num(backup_files_list); i++)
{
pgFile *file = (pgFile *) parray_get(backup_files_list, i);
/* if the entry was a directory, create it in the backup */
if (S_ISDIR(file->mode))
{
char dirpath[MAXPGPATH];
char *dir_name = GetRelativePath(file->path, pgdata);
if (verbose)
elog(LOG, "Create directory \"%s\"", dir_name);
join_path_components(dirpath, database_path, dir_name);
dir_create_dir(dirpath, DIR_PERMISSION);
}
__sync_lock_release(&file->lock);
}
/* sort by size for load balancing */
parray_qsort(backup_files_list, pgFileCompareSize);
/* init thread args with own file lists */
for (i = 0; i < num_threads; i++)
{
backup_files_args *arg = pg_malloc(sizeof(backup_files_args));
arg->from_root = pgdata;
arg->to_root = database_path;
arg->backup_files_list = backup_files_list;
arg->prev_backup_filelist = prev_backup_filelist;
arg->prev_backup_start_lsn = prev_backup_start_lsn;
backup_threads_args[i] = arg;
}
/* Run threads */
for (i = 0; i < num_threads; i++)
{
if (verbose)
elog(WARNING, "Start thread num:%li", parray_num(backup_threads_args[i]->backup_files_list));
pthread_create(&backup_threads[i], NULL, (void *(*)(void *)) backup_files, backup_threads_args[i]);
}
/* Wait theads */
for (i = 0; i < num_threads; i++)
{
pthread_join(backup_threads[i], NULL);
pg_free(backup_threads_args[i]);
}
/* clean previous backup file list */
if (prev_backup_filelist)
{
parray_walk(prev_backup_filelist, pgFileFree);
parray_free(prev_backup_filelist);
}
/* Notify end of backup */
pg_stop_backup(&current);
/* Add archived xlog files into the list of files of this backup */
if (stream_wal)
{
parray *xlog_files_list;
char pg_xlog_path[MAXPGPATH];
/* Wait for the completion of stream */
pthread_join(stream_thread, NULL);
/* Scan backup PG_XLOG_DIR */
xlog_files_list = parray_new();
join_path_components(pg_xlog_path, database_path, PG_XLOG_DIR);
dir_list_file(xlog_files_list, pg_xlog_path, false, true, false);
for (i = 0; i < parray_num(xlog_files_list); i++)
{
pgFile *file = (pgFile *) parray_get(xlog_files_list, i);
calc_file_checksum(file);
/* Remove file path root prefix*/
if (strstr(file->path, database_path) == file->path)
{
char *ptr = file->path;
file->path = pstrdup(GetRelativePath(ptr, database_path));
free(ptr);
}
}
/* Add xlog files into the list of backed up files */
parray_concat(backup_files_list, xlog_files_list);
parray_free(xlog_files_list);
}
/* Print the list of files to backup catalog */
write_backup_file_list(backup_files_list, pgdata);
/* Compute summary of size of regular files in the backup */
for (i = 0; i < parray_num(backup_files_list); i++)
{
pgFile *file = (pgFile *) parray_get(backup_files_list, i);
if (!S_ISREG(file->mode))
continue;
/* Count the amount of the data actually copied */
current.data_bytes += file->write_size;
}
if (backup_files_list)
parray_walk(backup_files_list, pgFileFree);
parray_free(backup_files_list);
}
/*
* Entry point of pg_probackup BACKUP subcommand.
*/
int
do_backup(void)
{
parray *backup_list;
bool is_ptrack_support;
/* PGDATA and BACKUP_MODE are always required */
if (pgdata == NULL)
elog(ERROR, "required parameter not specified: PGDATA "
"(-D, --pgdata)");
if (current.backup_mode == BACKUP_MODE_INVALID)
elog(ERROR, "required parameter not specified: BACKUP_MODE "
"(-b, --backup-mode)");
/* Create connection for PostgreSQL */
backup_conn = pgut_connect(pgut_dbname);
pgut_atexit_push(backup_disconnect, NULL);
/* Confirm that this server version is supported */
check_server_version();
/* Confirm data block size and xlog block size are compatible */
confirm_block_size("block_size", BLCKSZ);
confirm_block_size("wal_block_size", XLOG_BLCKSZ);
from_replica = pg_is_in_recovery();
current.checksum_version = get_data_checksum_version(true);
current.stream = stream_wal;
/* ptrack backup checks */
is_ptrack_support = pg_ptrack_support();
if (current.backup_mode == BACKUP_MODE_DIFF_PTRACK && !is_ptrack_support)
elog(ERROR, "This PostgreSQL instance does not support ptrack");
if (is_ptrack_support)
{
is_ptrack_enable = pg_ptrack_enable();
if(current.backup_mode == BACKUP_MODE_DIFF_PTRACK && !is_ptrack_enable)
elog(ERROR, "Ptrack is disabled");
}
/* Get exclusive lock of backup catalog */
catalog_lock();
/*
* Ensure that backup directory was initialized for the same PostgreSQL
* instance we opened connection to. And that target backup database PGDATA
* belogns to the same instance.
*/
check_system_identifiers();
/* get list of backups already taken */
backup_list = catalog_get_backup_list(INVALID_BACKUP_ID);
if (backup_list == NULL)
elog(ERROR, "Failed to get backup list.");
elog(LOG, "Backup start. backup-mode = %s, stream = %s",
pgBackupGetBackupMode(&current), current.stream ? "true" : "false");
/* Start backup. Update backup status. */
current.status = BACKUP_STATUS_RUNNING;
current.start_time = time(NULL);
/* Create backup directory and BACKUP_CONTROL_FILE */
if (pgBackupCreateDir(&current))
elog(ERROR, "cannot create backup directory");
pgBackupWriteBackupControlFile(&current);
elog(LOG, "Backup destination is initialized");
/* set the error processing function for the backup process */
pgut_atexit_push(backup_cleanup, NULL);
/* backup data */
do_backup_database(backup_list);
pgut_atexit_pop(backup_cleanup, NULL);
/* Backup is done. Update backup status */
current.end_time = time(NULL);
current.status = BACKUP_STATUS_DONE;
pgBackupWriteBackupControlFile(&current);
elog(LOG, "Backup completed. Total bytes : " INT64_FORMAT "",
current.data_bytes);
pgBackupValidate(&current);
elog(INFO, "Backup %s completed", base36enc(current.start_time));
/*
* After successfil backup completion remove backups
* which are expired according to retention policies
*/
if (delete_expired)
do_retention_purge();
return 0;
}
/*
* Confirm that this server version is supported
*/
static void
check_server_version(void)
{
static int server_version = 0;
/* confirm server version */
server_version = PQserverVersion(backup_conn);
if (server_version < 90500)
elog(ERROR,
"server version is %d.%d.%d, must be %s or higher",
server_version / 10000,
(server_version / 100) % 100,
server_version % 100, "9.5");
if (from_replica && server_version < 90600)
elog(ERROR,
"server version is %d.%d.%d, must be %s or higher for backup from replica",
server_version / 10000,
(server_version / 100) % 100,
server_version % 100, "9.6");
}
/*
* Ensure that backup directory was initialized for the same PostgreSQL
* instance we opened connection to. And that target backup database PGDATA
* belogns to the same instance.
* All system identifiers must be equal.
*/
static void
check_system_identifiers(void)
{
PGresult *res;
uint64 system_id_conn;
uint64 system_id_pgdata;
char *val;
system_id_pgdata = get_system_identifier();
res = pgut_execute(backup_conn,
"SELECT system_identifier FROM pg_control_system()",
0, NULL);
val = PQgetvalue(res, 0, 0);
PQclear(res);
if (!parse_uint64(val, &system_id_conn))
elog(ERROR, "%s is not system_identifier", val);
if (system_id_conn != system_identifier)
elog(ERROR, "Backup data directory was initialized for system id %ld, but connected instance system id is %ld",
system_identifier, system_id_conn);
if (system_id_pgdata != system_identifier)
elog(ERROR, "Backup data directory was initialized for system id %ld, but target backup directory system id is %ld",
system_identifier, system_id_pgdata);
}
/*
* Ensure that target backup database is initialized with
* compatible settings. Currently check BLCKSZ and XLOG_BLCKSZ.
*/
static void
confirm_block_size(const char *name, int blcksz)
{
PGresult *res;
char *endp;
int block_size;
res = pgut_execute(backup_conn, "SELECT current_setting($1)", 1, &name);
if (PQntuples(res) != 1 || PQnfields(res) != 1)
elog(ERROR, "cannot get %s: %s", name, PQerrorMessage(backup_conn));
block_size = strtol(PQgetvalue(res, 0, 0), &endp, 10);
PQclear(res);
if ((endp && *endp) || block_size != blcksz)
elog(ERROR,
"%s(%d) is not compatible(%d expected)",
name, block_size, blcksz);
}
/*
* Notify start of backup to PostgreSQL server.
*/
static void
pg_start_backup(const char *label, bool smooth, pgBackup *backup)
{
PGresult *res;
const char *params[2];
uint32 xlogid;
uint32 xrecoff;
params[0] = label;
/* 2nd argument is 'fast'*/
params[1] = smooth ? "false" : "true";
if (from_replica)
res = pgut_execute(backup_conn,
"SELECT pg_start_backup($1, $2, false)",
2,
params);
else
res = pgut_execute(backup_conn,
"SELECT pg_start_backup($1, $2)",
2,
params);
/* Extract timeline and LSN from results of pg_start_backup() */
XLogDataFromLSN(PQgetvalue(res, 0, 0), &xlogid, &xrecoff);
/* Calculate LSN */
backup->start_lsn = (XLogRecPtr) ((uint64) xlogid << 32) | xrecoff;
PQclear(res);
}
/*
* Switch to a new WAL segment. It should be called only for master.
*/
static void
pg_switch_wal(void)
{
PGresult *res;
/* Remove annoying NOTICE messages generated by backend */
res = pgut_execute(backup_conn, "SET client_min_messages = warning;", 0,
NULL);
PQclear(res);
#if PG_VERSION_NUM >= 100000
res = pgut_execute(backup_conn, "SELECT * FROM pg_switch_wal()", 0,
NULL);
#else
res = pgut_execute(backup_conn, "SELECT * FROM pg_switch_xlog()", 0,
NULL);
#endif
PQclear(res);
}
/*
* Check if the instance supports ptrack
* TODO Implement check of ptrack_version() instead of existing one
*/
static bool
pg_ptrack_support(void)
{
PGresult *res_db;
res_db = pgut_execute(backup_conn,
"SELECT proname FROM pg_proc WHERE proname='pg_ptrack_clear'",
0, NULL);
if (PQntuples(res_db) == 0)
{
PQclear(res_db);
return false;
}
PQclear(res_db);
return true;
}
/* Check if ptrack is enabled in target instance */
static bool
pg_ptrack_enable(void)
{
PGresult *res_db;
res_db = pgut_execute(backup_conn, "show ptrack_enable", 0, NULL);
if (strcmp(PQgetvalue(res_db, 0, 0), "on") != 0)
{
PQclear(res_db);
return false;
}
PQclear(res_db);
return true;
}
/* Check if target instance is replica */
static bool
pg_is_in_recovery(void)
{
PGresult *res_db;
res_db = pgut_execute(backup_conn, "SELECT pg_is_in_recovery()", 0, NULL);
if (PQgetvalue(res_db, 0, 0)[0] == 't')
{
PQclear(res_db);
return true;
}
PQclear(res_db);
return false;
}
/* Clear ptrack files in all databases of the instance we connected to */
static void
pg_ptrack_clear(void)
{
PGresult *res_db,
*res;
const char *dbname;
int i;
res_db = pgut_execute(backup_conn, "SELECT datname FROM pg_database",
0, NULL);
for(i = 0; i < PQntuples(res_db); i++)
{
PGconn *tmp_conn;
dbname = PQgetvalue(res_db, i, 0);
if (!strcmp(dbname, "template0"))
continue;
tmp_conn = pgut_connect(dbname);
res = pgut_execute(tmp_conn, "SELECT pg_ptrack_clear()", 0, NULL);
PQclear(res);
pgut_disconnect(tmp_conn);
}
PQclear(res_db);
}
/* Read and clear ptrack files of the target relation.
* Result is a bytea ptrack map of all segments of the target relation.
*/
static char *
pg_ptrack_get_and_clear(Oid tablespace_oid, Oid db_oid, Oid rel_oid,
size_t *result_size)
{
PGconn *tmp_conn;
PGresult *res_db,
*res;
char *dbname;
char *params[2];
char *result;
params[0] = palloc(64);
params[1] = palloc(64);
sprintf(params[0], "%i", db_oid);
res_db = pgut_execute(backup_conn,
"SELECT datname FROM pg_database WHERE oid=$1",
1, (const char **) params);
if (PQntuples(res_db) != 1 || PQnfields(res_db) != 1)
elog(ERROR, "cannot find database by oid %u", db_oid);
dbname = pstrdup(PQgetvalue(res_db, 0, 0));
PQclear(res_db);
tmp_conn = pgut_connect(dbname);
sprintf(params[0], "%i", tablespace_oid);
sprintf(params[1], "%i", rel_oid);
res = pgut_execute(tmp_conn, "SELECT pg_ptrack_get_and_clear($1, $2)",
2, (const char **)params);
if (PQnfields(res) != 1)
elog(ERROR, "cannot get ptrack file from database \"%s\" by tablespace oid %u and relation oid %u",
dbname, tablespace_oid, rel_oid);
result = (char *) PQunescapeBytea((unsigned char *) PQgetvalue(res, 0, 0),
result_size);
PQclear(res);
pgut_disconnect(tmp_conn);
pfree(params[0]);
pfree(params[1]);
pfree(dbname);
return result;
}
/*
* Wait for target 'lsn' to be archived in archive 'wal' directory with
* WAL segment file.
*/
static void
wait_archive_lsn(XLogRecPtr lsn, bool prev_segno)
{
TimeLineID tli;
XLogSegNo targetSegNo;
char wal_path[MAXPGPATH];
char wal_file[MAXFNAMELEN];
uint32 try_count = 0;
Assert(!stream_wal);
tli = get_current_timeline(false);
/* Compute the name of the WAL file containig requested LSN */
XLByteToSeg(lsn, targetSegNo);
if (prev_segno)
targetSegNo--;
XLogFileName(wal_file, tli, targetSegNo);
join_path_components(wal_path, arclog_path, wal_file);
/* Wait until switched WAL is archived */
while (!fileExists(wal_path))
{
sleep(1);
if (interrupted)
elog(ERROR, "interrupted during waiting for WAL archiving");
try_count++;
/* Inform user if WAL segment is absent in first attempt */
if (try_count == 1)
elog(INFO, "wait for LSN %X/%X in archived WAL segment %s",
(uint32) (lsn >> 32), (uint32) lsn, wal_path);
if (archive_timeout > 0 && try_count > archive_timeout)
elog(ERROR,
"switched WAL segment %s could not be archived in %d seconds",
wal_file, archive_timeout);
}
/*
* WAL segment was archived. Check LSN on it if we waited current WAL
* segment, not previous.
*/
if (!prev_segno && !wal_contains_lsn(arclog_path, lsn, tli))
elog(ERROR, "WAL segment %s doesn't contain target LSN %X/%X",
wal_file, (uint32) (lsn >> 32), (uint32) lsn);
}
/*
* Notify end of backup to PostgreSQL server.
*/
static void
pg_stop_backup(pgBackup *backup)
{
PGresult *res;
uint32 xlogid;
uint32 xrecoff;
/*
* We will use this values if there are no transactions between start_lsn
* and stop_lsn.
*/
time_t recovery_time;
TransactionId recovery_xid;
/* Remove annoying NOTICE messages generated by backend */
res = pgut_execute(backup_conn, "SET client_min_messages = warning;",
0, NULL);
PQclear(res);
if (from_replica)
/*
* Stop the non-exclusive backup. Besides stop_lsn it returns from
* pg_stop_backup(false) copy of the backup label and tablespace map
* so they can be written to disk by the caller.
*/
res = pgut_execute(backup_conn,
"SELECT *, txid_snapshot_xmax(txid_current_snapshot()),"
" current_timestamp(0)::timestamp"
" FROM pg_stop_backup(false)",
0, NULL);
else
res = pgut_execute(backup_conn,
"SELECT *, txid_snapshot_xmax(txid_current_snapshot()),"
" current_timestamp(0)::timestamp"
" FROM pg_stop_backup()",
0, NULL);
/* Extract timeline and LSN from results of pg_stop_backup() */
XLogDataFromLSN(PQgetvalue(res, 0, 0), &xlogid, &xrecoff);
/* Calculate LSN */
stop_backup_lsn = (XLogRecPtr) ((uint64) xlogid << 32) | xrecoff;
/* Write backup_label and tablespace_map for backup from replica */
if (from_replica)
{
char path[MAXPGPATH];
char backup_label[MAXPGPATH];
FILE *fp;
pgFile *file;
Assert(PQnfields(res) >= 5);
pgBackupGetPath(&current, path, lengthof(path), DATABASE_DIR);
/* Write backup_label */
join_path_components(backup_label, path, PG_BACKUP_LABEL_FILE);
fp = fopen(backup_label, "w");
if (fp == NULL)
elog(ERROR, "can't open backup label file \"%s\": %s",
backup_label, strerror(errno));
fwrite(PQgetvalue(res, 0, 1), 1, strlen(PQgetvalue(res, 0, 1)), fp);
fclose(fp);
/* TODO What for do we save the file into backup_list? */
file = pgFileNew(backup_label, true);
calc_file_checksum(file);
free(file->path);
file->path = strdup(PG_BACKUP_LABEL_FILE);
parray_append(backup_files_list, file);
/* Write tablespace_map */
if (strlen(PQgetvalue(res, 0, 2)) > 0)
{
char tablespace_map[MAXPGPATH];
join_path_components(tablespace_map, path, PG_TABLESPACE_MAP_FILE);
fp = fopen(tablespace_map, "w");
if (fp == NULL)
elog(ERROR, "can't open tablespace map file \"%s\": %s",
tablespace_map, strerror(errno));
fwrite(PQgetvalue(res, 0, 2), 1, strlen(PQgetvalue(res, 0, 2)), fp);
fclose(fp);
file = pgFileNew(tablespace_map, true);
calc_file_checksum(file);
free(file->path);
file->path = strdup(PG_TABLESPACE_MAP_FILE);
parray_append(backup_files_list, file);
}
if (sscanf(PQgetvalue(res, 0, 3), XID_FMT, &recovery_xid) != 1)
elog(ERROR,
"result of txid_snapshot_xmax() is invalid: %s",
PQerrorMessage(backup_conn));
if (!parse_time(PQgetvalue(res, 0, 4), &recovery_time))
elog(ERROR,
"result of current_timestamp is invalid: %s",
PQerrorMessage(backup_conn));
}
else
{
if (sscanf(PQgetvalue(res, 0, 1), XID_FMT, &recovery_xid) != 1)
elog(ERROR,
"result of txid_snapshot_xmax() is invalid: %s",
PQerrorMessage(backup_conn));
if (!parse_time(PQgetvalue(res, 0, 2), &recovery_time))
elog(ERROR,
"result of current_timestamp is invalid: %s",
PQerrorMessage(backup_conn));
}
PQclear(res);
if (!stream_wal)
wait_archive_lsn(stop_backup_lsn, false);
/* Fill in fields if that is the correct end of backup. */
if (backup != NULL)
{
char *xlog_path,
stream_xlog_path[MAXPGPATH];
if (stream_wal)
{
join_path_components(stream_xlog_path, pgdata, PG_XLOG_DIR);
xlog_path = stream_xlog_path;
}
else
xlog_path = arclog_path;
backup->tli = get_current_timeline(false);
backup->stop_lsn = stop_backup_lsn;
if (!read_recovery_info(xlog_path, backup->tli,
backup->start_lsn, backup->stop_lsn,
&backup->recovery_time, &backup->recovery_xid))
{
backup->recovery_time = recovery_time;
backup->recovery_xid = recovery_xid;
}
}
}
/*
* Return true if the path is a existing regular file.
*/
bool
fileExists(const char *path)
{
struct stat buf;
if (stat(path, &buf) == -1 && errno == ENOENT)
return false;
else if (!S_ISREG(buf.st_mode))
return false;
else
return true;
}
/*
* Notify end of backup to server when "backup_label" is in the root directory
* of the DB cluster.
* Also update backup status to ERROR when the backup is not finished.
*/
static void
backup_cleanup(bool fatal, void *userdata)
{
char path[MAXPGPATH];
/* If backup_label exists in $PGDATA, notify stop of backup to PostgreSQL */
join_path_components(path, pgdata, PG_BACKUP_LABEL_FILE);
if (fileExists(path))
{
elog(LOG, "%s exists, stop backup", PG_BACKUP_LABEL_FILE);
pg_stop_backup(NULL); /* don't care stop_lsn on error case */
}
/*
* Update status of backup in BACKUP_CONTROL_FILE to ERROR.
* end_time != 0 means backup finished
*/
if (current.status == BACKUP_STATUS_RUNNING && current.end_time == 0)
{
elog(LOG, "Backup is running, update its status to ERROR");
current.end_time = time(NULL);
current.status = BACKUP_STATUS_ERROR;
pgBackupWriteBackupControlFile(&current);
}
}
/*
* Disconnect backup connection during quit pg_probackup.
*/
static void
backup_disconnect(bool fatal, void *userdata)
{
pgut_disconnect(backup_conn);
}
/* Count bytes in file */
static long
file_size(const char *file_path)
{
long r;
FILE *f = fopen(file_path, "r");
if (!f)
{
elog(ERROR, "%s: cannot open file \"%s\" for reading: %s\n",
PROGRAM_NAME ,file_path, strerror(errno));
return -1;
}
fseek(f, 0, SEEK_END);
r = ftell(f);
fclose(f);
return r;
}
/*
* Find corresponding file in previous backup.
* Compare generations and return true if we don't need full copy
* of the file, but just part of it.
*
* skip_size - size of the file in previous backup. We can skip it
* and copy just remaining part of the file.
*/
bool
backup_compressed_file_partially(pgFile *file, void *arg, size_t *skip_size)
{
bool result = false;
pgFile *prev_file = NULL;
size_t current_file_size;
backup_files_args *arguments = (backup_files_args *) arg;
if (arguments->prev_backup_filelist)
{
pgFile **p = (pgFile **) parray_bsearch(arguments->prev_backup_filelist,
file, pgFileComparePath);
if (p)
prev_file = *p;
/* If file's gc generation has changed since last backup, just copy it*/
if (prev_file && prev_file->generation == file->generation)
{
current_file_size = file_size(file->path);
if (prev_file->write_size == BYTES_INVALID)
return false;
*skip_size = prev_file->write_size;
if (current_file_size >= prev_file->write_size)
{
elog(LOG, "Backup file %s partially: prev_size %lu, current_size %lu",
file->path, prev_file->write_size, current_file_size);
result = true;
}
else
elog(ERROR, "Something is wrong with %s. current_file_size %lu, prev %lu",
file->path, current_file_size, prev_file->write_size);
}
else
elog(LOG, "Copy full %s.", file->path);
}
return result;
}
/*
* Take a backup of the PGDATA at a file level.
* Copy all directories and files listed in backup_files_list.
* If the file is 'datafile' (regular relation's main fork), read it page by page,
* verify checksum and copy.
* In incremental backup mode, copy only files or datafiles' pages changed after
* previous backup.
* TODO review
*/
static void
backup_files(void *arg)
{
int i;
backup_files_args *arguments = (backup_files_args *) arg;
int n_backup_files_list = parray_num(arguments->backup_files_list);
/* backup a file or create a directory */
for (i = 0; i < n_backup_files_list; i++)
{
int ret;
struct stat buf;
pgFile *file = (pgFile *) parray_get(arguments->backup_files_list, i);
if (__sync_lock_test_and_set(&file->lock, 1) != 0)
continue;
/* check for interrupt */
if (interrupted)
elog(ERROR, "interrupted during backup");
if (progress)
elog(LOG, "Progress: (%d/%d). Process file \"%s\"",
i + 1, n_backup_files_list, file->path);
/* stat file to check its current state */
ret = stat(file->path, &buf);
if (ret == -1)
{
if (errno == ENOENT)
{
/*
* If file is not found, this is not en error.
* It could have been deleted by concurrent postgres transaction.
*/
file->write_size = BYTES_INVALID;
elog(LOG, "File \"%s\" is not found", file->path);
continue;
}
else
{
elog(ERROR,
"can't stat file to backup \"%s\": %s",
file->path, strerror(errno));
}
}
/* We have already copied all directories */
if (S_ISDIR(buf.st_mode))
continue;
if (S_ISREG(buf.st_mode))
{
/* skip files which have not been modified since last backup */
/* TODO Implement: compare oldfile and newfile checksum. Now it's just a stub */
if (arguments->prev_backup_filelist)
{
pgFile *prev_file = NULL;
pgFile **p = (pgFile **) parray_bsearch(arguments->prev_backup_filelist,
file, pgFileComparePath);
if (p)
prev_file = *p;
if (prev_file && false)
{
file->write_size = BYTES_INVALID;
if (verbose)
elog(LOG, "File \"%s\" has not changed since previous backup",
file->path);
continue;
}
}
/* copy the file into backup */
if (file->is_datafile)
{
if (is_compressed_data_file(file))
{
/* TODO review */
size_t skip_size = 0;
if (backup_compressed_file_partially(file, arguments, &skip_size))
{
/* backup cfs segment partly */
if (!copy_file_partly(arguments->from_root,
arguments->to_root,
file, skip_size))
{
/* record as skipped file in file_xxx.txt */
file->write_size = BYTES_INVALID;
elog(LOG, "skip");
continue;
}
}
else if (!copy_file(arguments->from_root,
arguments->to_root,
file))
{
/* record as skipped file in file_xxx.txt */
file->write_size = BYTES_INVALID;
elog(LOG, "skip");
continue;
}
}
else if (!backup_data_file(arguments->from_root,
arguments->to_root, file,
arguments->prev_backup_start_lsn))
{
file->write_size = BYTES_INVALID;
elog(LOG, "File \"%s\" was not copied to backup", file->path);
continue;
}
}
else if (!copy_file(arguments->from_root,
arguments->to_root,
file))
{
file->write_size = BYTES_INVALID;
elog(LOG, "File \"%s\" was not copied to backup", file->path);
continue;
}
elog(LOG, "File \"%s\". Copied %lu bytes",
file->path, (unsigned long) file->write_size);
}
else
elog(LOG, "unexpected file type %d", buf.st_mode);
}
}
/*
* Append files to the backup list array.
* TODO review
*/
static void
add_pgdata_files(parray *files, const char *root)
{
size_t i;
/* list files with the logical path. omit $PGDATA */
dir_list_file(files, root, true, true, false);
/* mark files that are possible datafile as 'datafile' */
for (i = 0; i < parray_num(files); i++)
{
pgFile *file = (pgFile *) parray_get(files, i);
char *relative;
char *fname;
size_t path_len;
/* data file must be a regular file */
if (!S_ISREG(file->mode))
continue;
/* data files are under "base", "global", or "pg_tblspc" */
relative = GetRelativePath(file->path, root);
if (!path_is_prefix_of_path("base", relative) &&
/*!path_is_prefix_of_path("global", relative) &&*/ //TODO What's wrong with this line?
!path_is_prefix_of_path(PG_TBLSPC_DIR, relative))
continue;
/* Get file name from path */
fname = last_dir_separator(relative);
if (fname == NULL)
fname = relative;
else
fname++;
/* Remove temp tables from the list */
if (fname[0] == 't' && isdigit(fname[1]))
{
pgFileFree(file);
parray_remove(files, i);
i--;
continue;
}
path_len = strlen(file->path);
/* Get link ptrack file to relations files */
if (path_len > 6 &&
strncmp(file->path + (path_len - 6), "ptrack", 6) == 0)
{
pgFile *search_file;
pgFile **pre_search_file;
int segno = 0;
while (true)
{
pgFile tmp_file;
tmp_file.path = pg_strdup(file->path);
/* Segno fits into 6 digits since it is not more than 4000 */
if (segno > 0)
sprintf(tmp_file.path + path_len - 7, ".%d", segno);
else
tmp_file.path[path_len - 7] = '\0';
pre_search_file = (pgFile **) parray_bsearch(files,
&tmp_file,
pgFileComparePath);
if (pre_search_file != NULL)
{
search_file = *pre_search_file;
search_file->ptrack_path = pg_strdup(file->path);
search_file->segno = segno;
}
else
{
pg_free(tmp_file.path);
break;
}
pg_free(tmp_file.path);
segno++;
}
/* Remove ptrack file itself from backup list */
pgFileFree(file);
parray_remove(files, i);
i--;
}
/* compress map file is not data file */
else if (path_len > 4 &&
strncmp(file->path + (path_len - 4), ".cfm", 4) == 0)
{
pgFile **pre_search_file;
pgFile tmp_file;
tmp_file.path = pg_strdup(file->path);
tmp_file.path[path_len - 4] = '\0';
pre_search_file = (pgFile **) parray_bsearch(files,
&tmp_file,
pgFileComparePath);
if (pre_search_file != NULL)
{
FileMap *map;
int md = open(file->path, O_RDWR|PG_BINARY, 0);
if (md < 0)
elog(ERROR, "cannot open cfm file '%s'", file->path);
map = cfs_mmap(md);
if (map == MAP_FAILED)
{
elog(LOG, "cfs_compression_ration failed to map file %s: %m", file->path);
close(md);
break;
}
(*pre_search_file)->generation = map->generation;
if (cfs_munmap(map) < 0)
elog(LOG, "CFS failed to unmap file %s: %m",
file->path);
if (close(md) < 0)
elog(LOG, "CFS failed to close file %s: %m",
file->path);
}
else
elog(ERROR, "corresponding segment '%s' is not found",
tmp_file.path);
pg_free(tmp_file.path);
}
/* name of data file start with digit */
else if (isdigit(fname[0]))
{
int find_dot;
int check_digit;
char *text_segno;
file->is_datafile = true;
/*
* Find segment number.
*/
for (find_dot = (int) path_len - 1;
file->path[find_dot] != '.' && find_dot >= 0;
find_dot--);
/* There is not segment number */
if (find_dot <= 0)
continue;
text_segno = file->path + find_dot + 1;
for (check_digit = 0; text_segno[check_digit] != '\0'; check_digit++)
if (!isdigit(text_segno[check_digit]))
{
check_digit = -1;
break;
}
if (check_digit != -1)
file->segno = (int) strtol(text_segno, NULL, 10);
}
}
}
/*
* Output the list of files to backup catalog DATABASE_FILE_LIST
*/
static void
write_backup_file_list(parray *files, const char *root)
{
FILE *fp;
char path[MAXPGPATH];
pgBackupGetPath(&current, path, lengthof(path), DATABASE_FILE_LIST);
fp = fopen(path, "wt");
if (fp == NULL)
elog(ERROR, "cannot open file list \"%s\": %s", path,
strerror(errno));
print_file_list(fp, files, root);
fclose(fp);
}
/*
* A helper function to create the path of a relation file and segment.
* The returned path is palloc'd
*/
static char *
datasegpath(RelFileNode rnode, ForkNumber forknum, BlockNumber segno)
{
char *path;
char *segpath;
path = relpathperm(rnode, forknum);
if (segno > 0)
{
segpath = psprintf("%s.%u", path, segno);
pfree(path);
return segpath;
}
else
return path;
}
/*
* Find pgfile by given rnode in the backup_files_list
* and add given blkno to its pagemap.
*/
void
process_block_change(ForkNumber forknum, RelFileNode rnode, BlockNumber blkno)
{
char *path;
char *rel_path;
BlockNumber blkno_inseg;
int segno;
pgFile *file_item = NULL;
int j;
segno = blkno / RELSEG_SIZE;
blkno_inseg = blkno % RELSEG_SIZE;
rel_path = datasegpath(rnode, forknum, segno);
path = pg_malloc(strlen(rel_path) + strlen(pgdata) + 2);
sprintf(path, "%s/%s", pgdata, rel_path);
for (j = 0; j < parray_num(backup_files_list); j++)
{
pgFile *p = (pgFile *) parray_get(backup_files_list, j);
if (strcmp(p->path, path) == 0)
{
file_item = p;
break;
}
}
/*
* If we don't have any record of this file in the file map, it means
* that it's a relation that did not have much activity since the last
* backup. We can safely ignore it. If it is a new relation file, the
* backup would simply copy it as-is.
*/
if (file_item)
datapagemap_add(&file_item->pagemap, blkno_inseg);
pg_free(path);
pg_free(rel_path);
}
/* TODO review it */
static void
make_pagemap_from_ptrack(parray *files)
{
size_t i;
for (i = 0; i < parray_num(files); i++)
{
pgFile *p = (pgFile *) parray_get(files, i);
if (p->ptrack_path != NULL)
{
char *tablespace;
Oid db_oid,
rel_oid,
tablespace_oid = 0;
int sep_iter,
sep_count = 0;
char *ptrack_nonparsed;
size_t ptrack_nonparsed_size = 0;
size_t start_addr;
tablespace = strstr(p->ptrack_path, PG_TBLSPC_DIR);
if (tablespace)
sscanf(tablespace+strlen(PG_TBLSPC_DIR)+1, "%i/", &tablespace_oid);
/*
* Path has format:
* pg_tblspc/tablespace_oid/tablespace_version_subdir/db_oid/rel_oid
* base/db_oid/rel_oid
*/
sep_iter = strlen(p->path);
while (sep_iter >= 0)
{
if (IS_DIR_SEP(p->path[sep_iter]))
sep_count++;
if (sep_count == 2)
break;
sep_iter--;
}
if (sep_iter <= 0)
elog(ERROR, "path of the file \"%s\" has wrong format",
p->path);
sscanf(p->path + sep_iter + 1, "%u/%u", &db_oid, &rel_oid);
ptrack_nonparsed = pg_ptrack_get_and_clear(tablespace_oid, db_oid,
rel_oid, &ptrack_nonparsed_size);
/* TODO What is 8? */
start_addr = (RELSEG_SIZE/8)*p->segno;
if (start_addr + RELSEG_SIZE/8 > ptrack_nonparsed_size)
p->pagemap.bitmapsize = ptrack_nonparsed_size - start_addr;
else
p->pagemap.bitmapsize = RELSEG_SIZE/8;
p->pagemap.bitmap = pg_malloc(p->pagemap.bitmapsize);
memcpy(p->pagemap.bitmap, ptrack_nonparsed+start_addr, p->pagemap.bitmapsize);
pg_free(ptrack_nonparsed);
}
}
}
/*
* Stop WAL streaming if current 'xlogpos' exceeds 'stop_backup_lsn', which is
* set by pg_stop_backup().
*/
static bool
stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
{
static uint32 prevtimeline = 0;
static XLogRecPtr prevpos = InvalidXLogRecPtr;
/* we assume that we get called once at the end of each segment */
if (verbose && segment_finished)
fprintf(stderr, _("%s: finished segment at %X/%X (timeline %u)\n"),
PROGRAM_NAME, (uint32) (xlogpos >> 32), (uint32) xlogpos,
timeline);
/*
* Note that we report the previous, not current, position here. After a
* timeline switch, xlogpos points to the beginning of the segment because
* that's where we always begin streaming. Reporting the end of previous
* timeline isn't totally accurate, because the next timeline can begin
* slightly before the end of the WAL that we received on the previous
* timeline, but it's close enough for reporting purposes.
*/
if (prevtimeline != 0 && prevtimeline != timeline)
fprintf(stderr, _("%s: switched to timeline %u at %X/%X\n"),
PROGRAM_NAME, timeline,
(uint32) (prevpos >> 32), (uint32) prevpos);
if (stop_backup_lsn != InvalidXLogRecPtr && xlogpos > stop_backup_lsn)
return true;
prevtimeline = timeline;
prevpos = xlogpos;
return false;
}
/*
* Start the log streaming
*/
static void
StreamLog(void *arg)
{
XLogRecPtr startpos;
TimeLineID starttli;
char *basedir = (char *)arg;
/*
* Connect in replication mode to the server
*/
if (conn == NULL)
conn = GetConnection();
if (!conn)
{
pthread_mutex_unlock(&check_stream_mut);
/* Error message already written in GetConnection() */
return;
}
if (!CheckServerVersionForStreaming(conn))
{
/*
* Error message already written in CheckServerVersionForStreaming().
* There's no hope of recovering from a version mismatch, so don't
* retry.
*/
disconnect_and_exit(1);
}
/*
* Identify server, obtaining start LSN position and current timeline ID
* at the same time, necessary if not valid data can be found in the
* existing output directory.
*/
if (!RunIdentifySystem(conn, NULL, &starttli, &startpos, NULL))
disconnect_and_exit(1);
/* Ok we have normal stream connect and main process can work again */
pthread_mutex_unlock(&check_stream_mut);
/*
* We must use startpos as start_lsn from start_backup
*/
startpos = current.start_lsn;
/*
* Always start streaming at the beginning of a segment
*/
startpos -= startpos % XLOG_SEG_SIZE;
/*
* Start the replication
*/
if (verbose)
fprintf(stderr,
_("%s: starting log streaming at %X/%X (timeline %u)\n"),
PROGRAM_NAME, (uint32) (startpos >> 32), (uint32) startpos,
starttli);
#if PG_VERSION_NUM >= 90600
{
StreamCtl ctl;
ctl.startpos = startpos;
ctl.timeline = starttli;
ctl.sysidentifier = NULL;
ctl.basedir = basedir;
ctl.stream_stop = stop_streaming;
ctl.standby_message_timeout = standby_message_timeout;
ctl.partial_suffix = NULL;
ctl.synchronous = false;
ctl.mark_done = false;
if(ReceiveXlogStream(conn, &ctl) == false)
elog(ERROR, "Problem in recivexlog");
}
#else
if(ReceiveXlogStream(conn, startpos, starttli, NULL, basedir,
stop_streaming, standby_message_timeout, NULL,
false, false) == false)
elog(ERROR, "Problem in recivexlog");
#endif
PQfinish(conn);
conn = NULL;
}
/*
* cfs_mmap() and cfs_munmap() function definitions mirror ones
* from cfs.h, but doesn't use atomic variables, since they are
* not allowed in frontend code.
* TODO Is it so?
* Since we cannot take atomic lock on files compressed by CFS,
* it should care about not changing files while backup is running.
*/
FileMap* cfs_mmap(int md)
{
FileMap* map;
#ifdef WIN32
HANDLE mh = CreateFileMapping(_get_osfhandle(md), NULL, PAGE_READWRITE,
0, (DWORD)sizeof(FileMap), NULL);
if (mh == NULL)
return (FileMap*)MAP_FAILED;
map = (FileMap*)MapViewOfFile(mh, FILE_MAP_ALL_ACCESS, 0, 0, 0);
CloseHandle(mh);
if (map == NULL)
return (FileMap*)MAP_FAILED;
#else
map = (FileMap*)mmap(NULL, sizeof(FileMap),
PROT_WRITE | PROT_READ, MAP_SHARED, md, 0);
#endif
return map;
}
int cfs_munmap(FileMap* map)
{
#ifdef WIN32
return UnmapViewOfFile(map) ? 0 : -1;
#else
return munmap(map, sizeof(FileMap));
#endif
}
Reference in New Issue View Git Blame Copy Permalink