db/pg_probackup
db/pg_probackup
1
0
Fork 0
mirror of https://github.com/postgrespro/pg_probackup.git synced 2025-02-08 14:28:36 +02:00
Code Issues Releases Activity

Merge remote-tracking branch 'origin/remote_pull'

Browse Source
This commit is contained in:
Arthur Zakirov 2019-04-16 13:24:02 +03:00
commit c440ab0f45
31 changed files with 2375 additions and 482 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Makefile
View File

@ -2,7 +2,7 @@ PROGRAM = pg_probackup
# utils
OBJS = src/utils/configuration.o src/utils/json.o src/utils/logger.o \
src/utils/parray.o src/utils/pgut.o src/utils/thread.o
src/utils/parray.o src/utils/pgut.o src/utils/thread.o src/utils/remote.o src/utils/file.o
OBJS += src/archive.o src/backup.o src/catalog.o src/configure.o src/data.o \
src/delete.o src/dir.o src/fetch.o src/help.o src/init.o src/merge.o \
@ -72,7 +72,6 @@ src/streamutil.h: $(top_srcdir)/src/bin/pg_basebackup/streamutil.h
src/xlogreader.c: $(top_srcdir)/src/backend/access/transam/xlogreader.c
rm -f $@ && $(LN_S) $(srchome)/src/backend/access/transam/xlogreader.c $@
ifeq (,$(filter 9.5 9.6,$(MAJORVERSION)))
src/walmethods.c: $(top_srcdir)/src/bin/pg_basebackup/walmethods.c
rm -f $@ && $(LN_S) $(srchome)/src/bin/pg_basebackup/walmethods.c $@

2
src/archive.c
View File

@ -55,7 +55,7 @@ do_archive_push(char *wal_file_path, char *wal_file_name, bool overwrite)
system_id);
/* Create 'archlog_path' directory. Do nothing if it already exists. */
dir_create_dir(arclog_path, DIR_PERMISSION);
fio_mkdir(arclog_path, DIR_PERMISSION, FIO_BACKUP_HOST);
join_path_components(absolute_wal_file_path, current_dir, wal_file_path);
join_path_components(backup_wal_file_path, arclog_path, wal_file_name);

78
src/backup.c
View File

@ -19,10 +19,12 @@
#include "streamutil.h"
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
#include "utils/thread.h"
#include <time.h>
#include "utils/file.h"
/*
* Macro needed to parse ptrack.
@ -497,7 +499,7 @@ do_backup_instance(void)
current.data_bytes = 0;
/* Obtain current timeline */
if (is_remote_backup)
if (IsReplicationProtocol())
{
char *sysidentifier;
TimeLineID starttli;
@ -545,7 +547,7 @@ do_backup_instance(void)
pgBackupGetPath(prev_backup, prev_backup_filelist_path,
lengthof(prev_backup_filelist_path), DATABASE_FILE_LIST);
/* Files of previous backup needed by DELTA backup */
prev_backup_filelist = dir_read_file_list(NULL, NULL, prev_backup_filelist_path);
prev_backup_filelist = dir_read_file_list(NULL, NULL, prev_backup_filelist_path, FIO_BACKUP_HOST);
/* If lsn is not NULL, only pages with higher lsn will be copied. */
prev_backup_start_lsn = prev_backup->start_lsn;
@ -603,7 +605,7 @@ do_backup_instance(void)
if (stream_wal)
{
join_path_components(dst_backup_path, database_path, PG_XLOG_DIR);
dir_create_dir(dst_backup_path, DIR_PERMISSION);
fio_mkdir(dst_backup_path, DIR_PERMISSION, FIO_BACKUP_HOST);
stream_thread_arg.basedir = dst_backup_path;
@ -637,7 +639,7 @@ do_backup_instance(void)
elog(ERROR, "Cannot continue backup because stream connect has failed.");
}
/* By default there are some error */
/* By default there are some error */
stream_thread_arg.ret = 1;
thread_interrupted = false;
@ -648,10 +650,12 @@ do_backup_instance(void)
backup_files_list = parray_new();
/* list files with the logical path. omit $PGDATA */
if (is_remote_backup)
if (IsReplicationProtocol())
get_remote_pgdata_filelist(backup_files_list);
else
dir_list_file(backup_files_list, instance_config.pgdata, true, true, false, 0);
dir_list_file(backup_files_list, instance_config.pgdata,
true, true, false, 0, FIO_DB_HOST);
/*
* Append to backup list all files and directories
@ -662,7 +666,7 @@ do_backup_instance(void)
/* External dirs numeration starts with 1.
* 0 value is not external dir */
dir_list_file(backup_files_list, parray_get(external_dirs, i),
false, true, false, i+1);
false, true, false, i+1, FIO_DB_HOST);
/* Sanity check for backup_files_list, thank you, Windows:
* https://github.com/postgrespro/pg_probackup/issues/48
@ -731,7 +735,7 @@ do_backup_instance(void)
char dirpath[MAXPGPATH];
char *dir_name;
if (!is_remote_backup)
if (!IsReplicationProtocol())
if (file->external_dir_num)
dir_name = GetRelativePath(file->path,
parray_get(external_dirs,
@ -752,7 +756,7 @@ do_backup_instance(void)
}
else
join_path_components(dirpath, database_path, dir_name);
dir_create_dir(dirpath, DIR_PERMISSION);
fio_mkdir(dirpath, DIR_PERMISSION, FIO_BACKUP_HOST);
}
/* setup threads */
@ -795,7 +799,7 @@ do_backup_instance(void)
elog(VERBOSE, "Start thread num: %i", i);
if (!is_remote_backup)
if (!IsReplicationProtocol())
pthread_create(&threads[i], NULL, backup_files, arg);
else
pthread_create(&threads[i], NULL, remote_backup_files, arg);
@ -865,20 +869,19 @@ do_backup_instance(void)
/* Add archived xlog files into the list of files of this backup */
if (stream_wal)
{
parray *xlog_files_list;
parray *xlog_files_list;
char pg_xlog_path[MAXPGPATH];
/* 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, 0);
dir_list_file(xlog_files_list, pg_xlog_path, false, true, false, 0, FIO_BACKUP_HOST);
for (i = 0; i < parray_num(xlog_files_list); i++)
{
pgFile *file = (pgFile *) parray_get(xlog_files_list, i);
if (S_ISREG(file->mode))
calc_file_checksum(file);
calc_file_checksum(file, FIO_BACKUP_HOST);
/* Remove file path root prefix*/
if (strstr(file->path, database_path) == file->path)
{
@ -888,7 +891,6 @@ do_backup_instance(void)
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);
@ -967,7 +969,7 @@ do_backup(time_t start_time, bool no_validate)
check_server_version();
/* TODO fix it for remote backup*/
if (!is_remote_backup)
if (!IsReplicationProtocol())
current.checksum_version = get_data_checksum_version(true);
is_checksum_enabled = pg_checksum_enable();
@ -1020,7 +1022,7 @@ do_backup(time_t start_time, bool no_validate)
* belogns to the same instance.
*/
/* TODO fix it for remote backup */
if (!is_remote_backup)
if (!IsReplicationProtocol())
check_system_identifiers();
/* Start backup. Update backup status. */
@ -1672,13 +1674,13 @@ wait_wal_lsn(XLogRecPtr lsn, bool is_start_lsn, bool wait_prev_segment)
{
if (!file_exists)
{
file_exists = fileExists(wal_segment_path);
file_exists = fileExists(wal_segment_path, FIO_BACKUP_HOST);
/* Try to find compressed WAL file */
if (!file_exists)
{
#ifdef HAVE_LIBZ
file_exists = fileExists(gz_wal_segment_path);
file_exists = fileExists(gz_wal_segment_path, FIO_BACKUP_HOST);
if (file_exists)
elog(LOG, "Found compressed WAL segment: %s", wal_segment_path);
#endif
@ -2071,16 +2073,15 @@ pg_stop_backup(pgBackup *backup)
/* Write backup_label */
join_path_components(backup_label, path, PG_BACKUP_LABEL_FILE);
fp = fopen(backup_label, PG_BINARY_W);
fp = fio_fopen(backup_label, PG_BINARY_W, FIO_BACKUP_HOST);
if (fp == NULL)
elog(ERROR, "can't open backup label file \"%s\": %s",
backup_label, strerror(errno));
len = strlen(PQgetvalue(res, 0, 3));
if (fwrite(PQgetvalue(res, 0, 3), 1, len, fp) != len ||
fflush(fp) != 0 ||
fsync(fileno(fp)) != 0 ||
fclose(fp))
if (fio_fwrite(fp, PQgetvalue(res, 0, 3), len) != len ||
fio_fflush(fp) != 0 ||
fio_fclose(fp))
elog(ERROR, "can't write backup label file \"%s\": %s",
backup_label, strerror(errno));
@ -2090,8 +2091,8 @@ pg_stop_backup(pgBackup *backup)
*/
if (backup_files_list)
{
file = pgFileNew(backup_label, true, 0);
calc_file_checksum(file);
file = pgFileNew(backup_label, true, 0, FIO_BACKUP_HOST);
calc_file_checksum(file, FIO_BACKUP_HOST);
free(file->path);
file->path = strdup(PG_BACKUP_LABEL_FILE);
parray_append(backup_files_list, file);
@ -2119,24 +2120,23 @@ pg_stop_backup(pgBackup *backup)
char tablespace_map[MAXPGPATH];
join_path_components(tablespace_map, path, PG_TABLESPACE_MAP_FILE);
fp = fopen(tablespace_map, PG_BINARY_W);
fp = fio_fopen(tablespace_map, PG_BINARY_W, FIO_BACKUP_HOST);
if (fp == NULL)
elog(ERROR, "can't open tablespace map file \"%s\": %s",
tablespace_map, strerror(errno));
len = strlen(val);
if (fwrite(val, 1, len, fp) != len ||
fflush(fp) != 0 ||
fsync(fileno(fp)) != 0 ||
fclose(fp))
if (fio_fwrite(fp, val, len) != len ||
fio_fflush(fp) != 0 ||
fio_fclose(fp))
elog(ERROR, "can't write tablespace map file \"%s\": %s",
tablespace_map, strerror(errno));
if (backup_files_list)
{
file = pgFileNew(tablespace_map, true, 0);
file = pgFileNew(tablespace_map, true, 0, FIO_BACKUP_HOST);
if (S_ISREG(file->mode))
calc_file_checksum(file);
calc_file_checksum(file, FIO_BACKUP_HOST);
free(file->path);
file->path = strdup(PG_TABLESPACE_MAP_FILE);
parray_append(backup_files_list, file);
@ -2307,7 +2307,7 @@ backup_files(void *arg)
i + 1, n_backup_files_list, file->path);
/* stat file to check its current state */
ret = stat(file->path, &buf);
ret = fio_stat(file->path, &buf, true, FIO_DB_HOST);
if (ret == -1)
{
if (errno == ENOENT)
@ -2382,7 +2382,8 @@ backup_files(void *arg)
}
else if (!file->external_dir_num &&
strcmp(file->name, "pg_control") == 0)
copy_pgcontrol_file(arguments->from_root, arguments->to_root,
copy_pgcontrol_file(arguments->from_root, FIO_DB_HOST,
arguments->to_root, FIO_BACKUP_HOST,
file);
else
{
@ -2395,7 +2396,7 @@ backup_files(void *arg)
if (prev_file && file->exists_in_prev &&
buf.st_mtime < current.parent_backup)
{
calc_file_checksum(file);
calc_file_checksum(file, FIO_DB_HOST);
/* ...and checksum is the same... */
if (EQ_TRADITIONAL_CRC32(file->crc, (*prev_file)->crc))
skip = true; /* ...skip copying file. */
@ -2414,7 +2415,8 @@ backup_files(void *arg)
src = arguments->from_root;
dst = arguments->to_root;
}
if (skip || !copy_file(src, dst, file))
if (skip ||
!copy_file(src, FIO_DB_HOST, dst, FIO_BACKUP_HOST, file))
{
/* disappeared file not to be confused with 'not changed' */
if (file->write_size != FILE_NOT_FOUND)

130
src/catalog.c
View File

@ -8,13 +8,13 @@
*-------------------------------------------------------------------------
*/
#include "pg_probackup.h"
#include <dirent.h>
#include <signal.h>
#include <sys/stat.h>
#include <unistd.h>
#include "pg_probackup.h"
#include "utils/file.h"
#include "utils/configuration.h"
static const char *backupModes[] = {"", "PAGE", "PTRACK", "DELTA", "FULL"};
@ -36,7 +36,7 @@ unlink_lock_atexit(void)
char *lock_file = (char *) parray_get(lock_files, i);
int res;
res = unlink(lock_file);
res = fio_unlink(lock_file, FIO_BACKUP_HOST);
if (res != 0 && errno != ENOENT)
elog(WARNING, "%s: %s", lock_file, strerror(errno));
}
@ -150,7 +150,7 @@ lock_backup(pgBackup *backup)
* Think not to make the file protection weaker than 0600. See
* comments below.
*/
fd = open(lock_file, O_RDWR | O_CREAT | O_EXCL, 0600);
fd = fio_open(lock_file, O_RDWR | O_CREAT | O_EXCL, FIO_BACKUP_HOST);
if (fd >= 0)
break; /* Success; exit the retry loop */
@ -165,7 +165,7 @@ lock_backup(pgBackup *backup)
* Read the file to get the old owner's PID. Note race condition
* here: file might have been deleted since we tried to create it.
*/
fd = open(lock_file, O_RDONLY, 0600);
fd = fio_open(lock_file, O_RDONLY, FIO_BACKUP_HOST);
if (fd < 0)
{
if (errno == ENOENT)
@ -173,10 +173,10 @@ lock_backup(pgBackup *backup)
elog(ERROR, "Could not open lock file \"%s\": %s",
lock_file, strerror(errno));
}
if ((len = read(fd, buffer, sizeof(buffer) - 1)) < 0)
if ((len = fio_read(fd, buffer, sizeof(buffer) - 1)) < 0)
elog(ERROR, "Could not read lock file \"%s\": %s",
lock_file, strerror(errno));
close(fd);
fio_close(fd);
if (len == 0)
elog(ERROR, "Lock file \"%s\" is empty", lock_file);
@ -221,7 +221,7 @@ lock_backup(pgBackup *backup)
* it. Need a loop because of possible race condition against other
* would-be creators.
*/
if (unlink(lock_file) < 0)
if (fio_unlink(lock_file, FIO_BACKUP_HOST) < 0)
elog(ERROR, "Could not remove old lock file \"%s\": %s",
lock_file, strerror(errno));
}
@ -232,32 +232,32 @@ lock_backup(pgBackup *backup)
snprintf(buffer, sizeof(buffer), "%d\n", my_pid);
errno = 0;
if (write(fd, buffer, strlen(buffer)) != strlen(buffer))
if (fio_write(fd, buffer, strlen(buffer)) != strlen(buffer))
{
int save_errno = errno;
close(fd);
unlink(lock_file);
fio_close(fd);
fio_unlink(lock_file, FIO_BACKUP_HOST);
/* if write didn't set errno, assume problem is no disk space */
errno = save_errno ? save_errno : ENOSPC;
elog(ERROR, "Could not write lock file \"%s\": %s",
lock_file, strerror(errno));
}
if (fsync(fd) != 0)
if (fio_flush(fd) != 0)
{
int save_errno = errno;
close(fd);
unlink(lock_file);
fio_close(fd);
fio_unlink(lock_file, FIO_BACKUP_HOST);
errno = save_errno;
elog(ERROR, "Could not write lock file \"%s\": %s",
lock_file, strerror(errno));
}
if (close(fd) != 0)
if (fio_close(fd) != 0)
{
int save_errno = errno;
unlink(lock_file);
fio_unlink(lock_file, FIO_BACKUP_HOST);
errno = save_errno;
elog(ERROR, "Culd not write lock file \"%s\": %s",
lock_file, strerror(errno));
@ -290,14 +290,14 @@ pgBackupGetBackupMode(pgBackup *backup)
}
static bool
IsDir(const char *dirpath, const char *entry)
IsDir(const char *dirpath, const char *entry, fio_location location)
{
char path[MAXPGPATH];
struct stat st;
snprintf(path, MAXPGPATH, "%s/%s", dirpath, entry);
return stat(path, &st) == 0 && S_ISDIR(st.st_mode);
return fio_stat(path, &st, false, location) == 0 && S_ISDIR(st.st_mode);
}
/*
@ -315,7 +315,7 @@ catalog_get_backup_list(time_t requested_backup_id)
int i;
/* open backup instance backups directory */
data_dir = opendir(backup_instance_path);
data_dir = fio_opendir(backup_instance_path, FIO_BACKUP_HOST);
if (data_dir == NULL)
{
elog(WARNING, "cannot open directory \"%s\": %s", backup_instance_path,
@ -325,14 +325,14 @@ catalog_get_backup_list(time_t requested_backup_id)
/* scan the directory and list backups */
backups = parray_new();
for (; (data_ent = readdir(data_dir)) != NULL; errno = 0)
for (; (data_ent = fio_readdir(data_dir)) != NULL; errno = 0)
{
char backup_conf_path[MAXPGPATH];
char data_path[MAXPGPATH];
pgBackup *backup = NULL;
/* skip not-directory entries and hidden entries */
if (!IsDir(backup_instance_path, data_ent->d_name)
if (!IsDir(backup_instance_path, data_ent->d_name, FIO_BACKUP_HOST)
|| data_ent->d_name[0] == '.')
continue;
@ -378,7 +378,7 @@ catalog_get_backup_list(time_t requested_backup_id)
goto err_proc;
}
closedir(data_dir);
fio_closedir(data_dir);
data_dir = NULL;
parray_qsort(backups, pgBackupCompareIdDesc);
@ -404,7 +404,7 @@ catalog_get_backup_list(time_t requested_backup_id)
err_proc:
if (data_dir)
closedir(data_dir);
fio_closedir(data_dir);
if (backups)
parray_walk(backups, pgBackupFree);
parray_free(backups);
@ -492,13 +492,13 @@ pgBackupCreateDir(pgBackup *backup)
if (!dir_is_empty(path))
elog(ERROR, "backup destination is not empty \"%s\"", path);
dir_create_dir(path, DIR_PERMISSION);
fio_mkdir(path, DIR_PERMISSION, FIO_BACKUP_HOST);
/* create directories for actual backup files */
for (i = 0; i < parray_num(subdirs); i++)
{
pgBackupGetPath(backup, path, lengthof(path), parray_get(subdirs, i));
dir_create_dir(path, DIR_PERMISSION);
fio_mkdir(path, DIR_PERMISSION, FIO_BACKUP_HOST);
}
free_dir_list(subdirs);
@ -513,51 +513,51 @@ pgBackupWriteControl(FILE *out, pgBackup *backup)
{
char timestamp[100];
fprintf(out, "#Configuration\n");
fprintf(out, "backup-mode = %s\n", pgBackupGetBackupMode(backup));
fprintf(out, "stream = %s\n", backup->stream ? "true" : "false");
fprintf(out, "compress-alg = %s\n",
fio_fprintf(out, "#Configuration\n");
fio_fprintf(out, "backup-mode = %s\n", pgBackupGetBackupMode(backup));
fio_fprintf(out, "stream = %s\n", backup->stream ? "true" : "false");
fio_fprintf(out, "compress-alg = %s\n",
deparse_compress_alg(backup->compress_alg));
fprintf(out, "compress-level = %d\n", backup->compress_level);
fprintf(out, "from-replica = %s\n", backup->from_replica ? "true" : "false");
fio_fprintf(out, "compress-level = %d\n", backup->compress_level);
fio_fprintf(out, "from-replica = %s\n", backup->from_replica ? "true" : "false");
fprintf(out, "\n#Compatibility\n");
fprintf(out, "block-size = %u\n", backup->block_size);
fprintf(out, "xlog-block-size = %u\n", backup->wal_block_size);
fprintf(out, "checksum-version = %u\n", backup->checksum_version);
fio_fprintf(out, "\n#Compatibility\n");
fio_fprintf(out, "block-size = %u\n", backup->block_size);
fio_fprintf(out, "xlog-block-size = %u\n", backup->wal_block_size);
fio_fprintf(out, "checksum-version = %u\n", backup->checksum_version);
if (backup->program_version[0] != '\0')
fprintf(out, "program-version = %s\n", backup->program_version);
fio_fprintf(out, "program-version = %s\n", backup->program_version);
if (backup->server_version[0] != '\0')
fprintf(out, "server-version = %s\n", backup->server_version);
fio_fprintf(out, "server-version = %s\n", backup->server_version);
fprintf(out, "\n#Result backup info\n");
fprintf(out, "timelineid = %d\n", backup->tli);
fio_fprintf(out, "\n#Result backup info\n");
fio_fprintf(out, "timelineid = %d\n", backup->tli);
/* LSN returned by pg_start_backup */
fprintf(out, "start-lsn = %X/%X\n",
fio_fprintf(out, "start-lsn = %X/%X\n",
(uint32) (backup->start_lsn >> 32),
(uint32) backup->start_lsn);
/* LSN returned by pg_stop_backup */
fprintf(out, "stop-lsn = %X/%X\n",
fio_fprintf(out, "stop-lsn = %X/%X\n",
(uint32) (backup->stop_lsn >> 32),
(uint32) backup->stop_lsn);
time2iso(timestamp, lengthof(timestamp), backup->start_time);
fprintf(out, "start-time = '%s'\n", timestamp);
fio_fprintf(out, "start-time = '%s'\n", timestamp);
if (backup->merge_time > 0)
{
time2iso(timestamp, lengthof(timestamp), backup->merge_time);
fprintf(out, "merge-time = '%s'\n", timestamp);
fio_fprintf(out, "merge-time = '%s'\n", timestamp);
}
if (backup->end_time > 0)
{
time2iso(timestamp, lengthof(timestamp), backup->end_time);
fprintf(out, "end-time = '%s'\n", timestamp);
fio_fprintf(out, "end-time = '%s'\n", timestamp);
}
fprintf(out, "recovery-xid = " XID_FMT "\n", backup->recovery_xid);
fio_fprintf(out, "recovery-xid = " XID_FMT "\n", backup->recovery_xid);
if (backup->recovery_time > 0)
{
time2iso(timestamp, lengthof(timestamp), backup->recovery_time);
fprintf(out, "recovery-time = '%s'\n", timestamp);
fio_fprintf(out, "recovery-time = '%s'\n", timestamp);
}
/*
@ -565,24 +565,24 @@ pgBackupWriteControl(FILE *out, pgBackup *backup)
* WAL segments in archive 'wal' directory.
*/
if (backup->data_bytes != BYTES_INVALID)
fprintf(out, "data-bytes = " INT64_FORMAT "\n", backup->data_bytes);
fio_fprintf(out, "data-bytes = " INT64_FORMAT "\n", backup->data_bytes);
if (backup->wal_bytes != BYTES_INVALID)
fprintf(out, "wal-bytes = " INT64_FORMAT "\n", backup->wal_bytes);
fio_fprintf(out, "wal-bytes = " INT64_FORMAT "\n", backup->wal_bytes);
fprintf(out, "status = %s\n", status2str(backup->status));
fio_fprintf(out, "status = %s\n", status2str(backup->status));
/* 'parent_backup' is set if it is incremental backup */
if (backup->parent_backup != 0)
fprintf(out, "parent-backup-id = '%s'\n", base36enc(backup->parent_backup));
fio_fprintf(out, "parent-backup-id = '%s'\n", base36enc(backup->parent_backup));
/* print connection info except password */
if (backup->primary_conninfo)
fprintf(out, "primary_conninfo = '%s'\n", backup->primary_conninfo);
fio_fprintf(out, "primary_conninfo = '%s'\n", backup->primary_conninfo);
/* print external directories list */
if (backup->external_dir_str)
fprintf(out, "external-dirs = '%s'\n", backup->external_dir_str);
fio_fprintf(out, "external-dirs = '%s'\n", backup->external_dir_str);
}
/*
@ -599,27 +599,25 @@ write_backup(pgBackup *backup)
pgBackupGetPath(backup, path, lengthof(path), BACKUP_CONTROL_FILE);
snprintf(path_temp, sizeof(path_temp), "%s.tmp", path);
fp = fopen(path_temp, "wt");
fp = fio_fopen(path_temp, PG_BINARY_W, FIO_BACKUP_HOST);
if (fp == NULL)
elog(ERROR, "Cannot open configuration file \"%s\": %s",
path_temp, strerror(errno));
pgBackupWriteControl(fp, backup);
if (fflush(fp) != 0 ||
fsync(fileno(fp)) != 0 ||
fclose(fp))
if (fio_fflush(fp) || fio_fclose(fp))
{
errno_temp = errno;
unlink(path_temp);
fio_unlink(path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot write configuration file \"%s\": %s",
path_temp, strerror(errno_temp));
}
if (rename(path_temp, path) < 0)
if (fio_rename(path_temp, path, FIO_BACKUP_HOST) < 0)
{
errno_temp = errno;
unlink(path_temp);
fio_unlink(path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot rename configuration file \"%s\" to \"%s\": %s",
path_temp, path, strerror(errno_temp));
}
@ -640,27 +638,25 @@ write_backup_filelist(pgBackup *backup, parray *files, const char *root,
pgBackupGetPath(backup, path, lengthof(path), DATABASE_FILE_LIST);
snprintf(path_temp, sizeof(path_temp), "%s.tmp", path);
fp = fopen(path_temp, "wt");
fp = fio_fopen(path_temp, PG_BINARY_W, FIO_BACKUP_HOST);
if (fp == NULL)
elog(ERROR, "Cannot open file list \"%s\": %s", path_temp,
strerror(errno));
print_file_list(fp, files, root, external_prefix, external_list);
if (fflush(fp) != 0 ||
fsync(fileno(fp)) != 0 ||
fclose(fp))
if (fio_fflush(fp) || fio_fclose(fp))
{
errno_temp = errno;
unlink(path_temp);
fio_unlink(path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot write file list \"%s\": %s",
path_temp, strerror(errno));
}
if (rename(path_temp, path) < 0)
if (fio_rename(path_temp, path, FIO_BACKUP_HOST) < 0)
{
errno_temp = errno;
unlink(path_temp);
fio_unlink(path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot rename configuration file \"%s\" to \"%s\": %s",
path_temp, path, strerror(errno_temp));
}
@ -715,7 +711,7 @@ readBackupControlFile(const char *path)
};
pgBackupInit(backup);
if (access(path, F_OK) != 0)
if (fio_access(path, F_OK, FIO_BACKUP_HOST) != 0)
{
elog(WARNING, "Control file \"%s\" doesn't exist", path);
pgBackupFree(backup);

46
src/configure.c
View File

@ -39,6 +39,7 @@ static void show_configure_json(ConfigOption *opt);
#define OPTION_LOG_GROUP "Logging parameters"
#define OPTION_RETENTION_GROUP "Retention parameters"
#define OPTION_COMPRESS_GROUP "Compression parameters"
#define OPTION_REMOTE_GROUP "Remote access parameters"
/*
* Short name should be non-printable ASCII character.
@ -179,6 +180,42 @@ ConfigOption instance_options[] =
&instance_config.compress_level, SOURCE_CMD, 0,
OPTION_COMPRESS_GROUP, 0, option_get_value
},
/* Remote backup options */
{
's', 219, "remote-proto",
&instance_config.remote.proto, SOURCE_CMD, 0,
OPTION_REMOTE_GROUP, 0, option_get_value
},
{
's', 220, "remote-host",
&instance_config.remote.host, SOURCE_CMD, 0,
OPTION_REMOTE_GROUP, 0, option_get_value
},
{
's', 221, "remote-port",
&instance_config.remote.port, SOURCE_CMD, 0,
OPTION_REMOTE_GROUP, 0, option_get_value
},
{
's', 222, "remote-path",
&instance_config.remote.path, SOURCE_CMD, 0,
OPTION_REMOTE_GROUP, 0, option_get_value
},
{
's', 223, "remote-user",
&instance_config.remote.user, SOURCE_CMD, 0,
OPTION_REMOTE_GROUP, 0, option_get_value
},
{
's', 224, "ssh-options",
&instance_config.remote.ssh_options, SOURCE_CMD, 0,
OPTION_REMOTE_GROUP, 0, option_get_value
},
{
's', 225, "ssh-config",
&instance_config.remote.ssh_config, SOURCE_CMD, 0,
OPTION_REMOTE_GROUP, 0, option_get_value
},
{ 0 }
};
@ -225,7 +262,7 @@ do_set_config(bool missing_ok)
join_path_components(path, backup_instance_path, BACKUP_CATALOG_CONF_FILE);
snprintf(path_temp, sizeof(path_temp), "%s.tmp", path);
if (!missing_ok && !fileExists(path))
if (!missing_ok && !fileExists(path, FIO_LOCAL_HOST))
elog(ERROR, "Configuration file \"%s\" doesn't exist", path);
fp = fopen(path_temp, "wt");
@ -256,7 +293,10 @@ do_set_config(bool missing_ok)
fprintf(fp, "# %s\n", current_group);
}
fprintf(fp, "%s = %s\n", opt->lname, value);
if (strchr(value, ' '))
fprintf(fp, "%s = '%s'\n", opt->lname, value);
else
fprintf(fp, "%s = %s\n", opt->lname, value);
pfree(value);
}
@ -298,6 +338,8 @@ init_config(InstanceConfig *config)
config->compress_alg = COMPRESS_ALG_DEFAULT;
config->compress_level = COMPRESS_LEVEL_DEFAULT;
config->remote.proto = (char*)"ssh";
}
static void

370
src/data.c
View File

@ -13,9 +13,9 @@
#include "storage/checksum.h"
#include "storage/checksum_impl.h"
#include <common/pg_lzcompress.h>
#include "utils/file.h"
#include <unistd.h>
#include <sys/stat.h>
#ifdef HAVE_LIBZ
@ -62,7 +62,7 @@ zlib_decompress(void *dst, size_t dst_size, void const *src, size_t src_size)
* Compresses source into dest using algorithm. Returns the number of bytes
* written in the destination buffer, or -1 if compression fails.
*/
static int32
int32
do_compress(void* dst, size_t dst_size, void const* src, size_t src_size,
CompressAlg alg, int level, const char **errormsg)
{
@ -168,22 +168,8 @@ page_may_be_compressed(Page page, CompressAlg alg, uint32 backup_version)
return false;
}
/*
* When copying datafiles to backup we validate and compress them block
* by block. Thus special header is required for each data block.
*/
typedef struct BackupPageHeader
{
BlockNumber block; /* block number */
int32 compressed_size;
} BackupPageHeader;
/* Special value for compressed_size field */
#define PageIsTruncated -2
#define SkipCurrentPage -3
/* Verify page's header */
static bool
bool
parse_page(Page page, XLogRecPtr *lsn)
{
PageHeader phdr = (PageHeader) page;
@ -215,14 +201,10 @@ read_page_from_file(pgFile *file, BlockNumber blknum,
FILE *in, Page page, XLogRecPtr *page_lsn)
{
off_t offset = blknum * BLCKSZ;
size_t read_len = 0;
ssize_t read_len = 0;
/* read the block */
if (fseek(in, offset, SEEK_SET) != 0)
elog(ERROR, "File: %s, could not seek to block %u: %s",
file->path, blknum, strerror(errno));
read_len = fread(page, 1, BLCKSZ, in);
read_len = fio_pread(in, page, offset);
if (read_len != BLCKSZ)
{
@ -234,9 +216,12 @@ read_page_from_file(pgFile *file, BlockNumber blknum,
return 0;
}
else
{
elog(WARNING, "File: %s, block %u, expected block size %u,"
"but read %zu, try again",
file->path, blknum, BLCKSZ, read_len);
return -1;
}
}
/*
@ -269,14 +254,14 @@ read_page_from_file(pgFile *file, BlockNumber blknum,
}
/* Verify checksum */
if(current.checksum_version)
if (current.checksum_version)
{
BlockNumber blkno = file->segno * RELSEG_SIZE + blknum;
/*
* If checksum is wrong, sleep a bit and then try again
* several times. If it didn't help, throw error
*/
if (pg_checksum_page(page, file->segno * RELSEG_SIZE + blknum)
!= ((PageHeader) page)->pd_checksum)
if (pg_checksum_page(page, blkno) != ((PageHeader) page)->pd_checksum)
{
elog(WARNING, "File: %s blknum %u have wrong checksum, try again",
file->path, blknum);
@ -309,7 +294,7 @@ static int32
prepare_page(backup_files_arg *arguments,
pgFile *file, XLogRecPtr prev_backup_start_lsn,
BlockNumber blknum, BlockNumber nblocks,
FILE *in, int *n_skipped,
FILE *in, BlockNumber *n_skipped,
BackupMode backup_mode,
Page page)
{
@ -332,8 +317,7 @@ prepare_page(backup_files_arg *arguments,
{
while(!page_is_valid && try_again)
{
int result = read_page_from_file(file, blknum,
in, page, &page_lsn);
int result = read_page_from_file(file, blknum, in, page, &page_lsn);
try_again--;
if (result == 0)
@ -494,12 +478,12 @@ compress_and_backup_page(pgFile *file, BlockNumber blknum,
COMP_FILE_CRC32(true, *crc, write_buffer, write_buffer_size);
/* write data page */
if(fwrite(write_buffer, 1, write_buffer_size, out) != write_buffer_size)
if (fio_fwrite(out, write_buffer, write_buffer_size) != write_buffer_size)
{
int errno_tmp = errno;
fclose(in);
fclose(out);
fio_fclose(out);
elog(ERROR, "File: %s, cannot write backup at block %u: %s",
file->path, blknum, strerror(errno_tmp));
}
@ -525,8 +509,8 @@ backup_data_file(backup_files_arg* arguments,
FILE *out;
BlockNumber blknum = 0;
BlockNumber nblocks = 0;
int n_blocks_skipped = 0;
int n_blocks_read = 0;
BlockNumber n_blocks_skipped = 0;
BlockNumber n_blocks_read = 0;
int page_state;
char curr_page[BLCKSZ];
@ -554,7 +538,7 @@ backup_data_file(backup_files_arg* arguments,
INIT_FILE_CRC32(true, file->crc);
/* open backup mode file for read */
in = fopen(file->path, PG_BINARY_R);
in = fio_fopen(file->path, PG_BINARY_R, FIO_DB_HOST);
if (in == NULL)
{
FIN_FILE_CRC32(true, file->crc);
@ -576,7 +560,7 @@ backup_data_file(backup_files_arg* arguments,
if (file->size % BLCKSZ != 0)
{
fclose(in);
fio_fclose(in);
elog(WARNING, "File: %s, invalid file size %zu", file->path, file->size);
}
@ -588,11 +572,11 @@ backup_data_file(backup_files_arg* arguments,
nblocks = file->size/BLCKSZ;
/* open backup file for write */
out = fopen(to_path, PG_BINARY_W);
out = fio_fopen(to_path, PG_BINARY_W, FIO_BACKUP_HOST);
if (out == NULL)
{
int errno_tmp = errno;
fclose(in);
fio_fclose(in);
elog(ERROR, "cannot open backup file \"%s\": %s",
to_path, strerror(errno_tmp));
}
@ -607,16 +591,32 @@ backup_data_file(backup_files_arg* arguments,
if (file->pagemap.bitmapsize == PageBitmapIsEmpty ||
file->pagemap_isabsent || !file->exists_in_prev)
{
for (blknum = 0; blknum < nblocks; blknum++)
if (backup_mode != BACKUP_MODE_DIFF_PTRACK && fio_is_remote_file(in))
{
page_state = prepare_page(arguments, file, prev_backup_start_lsn,
blknum, nblocks, in, &n_blocks_skipped,
backup_mode, curr_page);
compress_and_backup_page(file, blknum, in, out, &(file->crc),
page_state, curr_page, calg, clevel);
n_blocks_read++;
if (page_state == PageIsTruncated)
break;
int rc = fio_send_pages(in, out, file,
backup_mode == BACKUP_MODE_DIFF_DELTA && file->exists_in_prev ? prev_backup_start_lsn : InvalidXLogRecPtr,
&n_blocks_skipped, calg, clevel);
if (rc == PAGE_CHECKSUM_MISMATCH && is_ptrack_support)
goto RetryUsingPtrack;
if (rc < 0)
elog(ERROR, "Failed to read file %s: %s",
file->path, rc == PAGE_CHECKSUM_MISMATCH ? "data file checksum mismatch" : strerror(-rc));
n_blocks_read = rc;
}
else
{
RetryUsingPtrack:
for (blknum = 0; blknum < nblocks; blknum++)
{
page_state = prepare_page(arguments, file, prev_backup_start_lsn,
blknum, nblocks, in, &n_blocks_skipped,
backup_mode, curr_page);
compress_and_backup_page(file, blknum, in, out, &(file->crc),
page_state, curr_page, calg, clevel);
n_blocks_read++;
if (page_state == PageIsTruncated)
break;
}
}
if (backup_mode == BACKUP_MODE_DIFF_DELTA)
file->n_blocks = n_blocks_read;
@ -647,21 +647,20 @@ backup_data_file(backup_files_arg* arguments,
}
/* update file permission */
if (chmod(to_path, FILE_PERMISSION) == -1)
if (fio_chmod(to_path, FILE_PERMISSION, FIO_BACKUP_HOST) == -1)
{
int errno_tmp = errno;
fclose(in);
fclose(out);
fio_fclose(in);
fio_fclose(out);
elog(ERROR, "cannot change mode of \"%s\": %s", file->path,
strerror(errno_tmp));
}
if (fflush(out) != 0 ||
fsync(fileno(out)) != 0 ||
fclose(out))
if (fio_fflush(out) != 0 ||
fio_fclose(out))
elog(ERROR, "cannot write backup file \"%s\": %s",
to_path, strerror(errno));
fclose(in);
fio_fclose(in);
FIN_FILE_CRC32(true, file->crc);
@ -671,7 +670,7 @@ backup_data_file(backup_files_arg* arguments,
*/
if (n_blocks_read != 0 && n_blocks_read == n_blocks_skipped)
{
if (remove(to_path) == -1)
if (fio_unlink(to_path, FIO_BACKUP_HOST) == -1)
elog(ERROR, "cannot remove file \"%s\": %s", to_path,
strerror(errno));
return false;
@ -715,9 +714,7 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
* modified pages for differential restore. If the file does not exist,
* re-open it with "w" to create an empty file.
*/
out = fopen(to_path, PG_BINARY_R "+");
if (out == NULL && errno == ENOENT)
out = fopen(to_path, PG_BINARY_W);
out = fio_fopen(to_path, PG_BINARY_R "+", FIO_DB_HOST);
if (out == NULL)
{
int errno_tmp = errno;
@ -832,7 +829,7 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
/*
* Seek and write the restored page.
*/
if (fseek(out, write_pos, SEEK_SET) < 0)
if (fio_fseek(out, write_pos) < 0)
elog(ERROR, "Cannot seek block %u of \"%s\": %s",
blknum, to_path, strerror(errno));
@ -840,7 +837,7 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
{
/* We uncompressed the page, so its size is BLCKSZ */
header.compressed_size = BLCKSZ;
if (fwrite(&header, 1, sizeof(header), out) != sizeof(header))
if (fio_fwrite(out, &header, sizeof(header)) != sizeof(header))
elog(ERROR, "Cannot write header of block %u of \"%s\": %s",
blknum, file->path, strerror(errno));
}
@ -851,14 +848,13 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
*/
if (uncompressed_size == BLCKSZ)
{
if (fwrite(page.data, 1, BLCKSZ, out) != BLCKSZ)
if (fio_fwrite(out, page.data, BLCKSZ) != BLCKSZ)
elog(ERROR, "Cannot write block %u of \"%s\": %s",
blknum, file->path, strerror(errno));
}
else
{
/* */
if (fwrite(compressed_page.data, 1, BLCKSZ, out) != BLCKSZ)
if (fio_fwrite(out, compressed_page.data, BLCKSZ) != BLCKSZ)
elog(ERROR, "Cannot write block %u of \"%s\": %s",
blknum, file->path, strerror(errno));
}
@ -873,13 +869,8 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
*/
if (allow_truncate && file->n_blocks != BLOCKNUM_INVALID && !need_truncate)
{
size_t file_size = 0;
/* get file current size */
fseek(out, 0, SEEK_END);
file_size = ftell(out);
if (file_size > file->n_blocks * BLCKSZ)
struct stat st;
if (fio_ffstat(out, &st) == 0 && st.st_size > file->n_blocks * BLCKSZ)
{
truncate_from = file->n_blocks;
need_truncate = true;
@ -896,7 +887,7 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
/*
* Truncate file to this length.
*/
if (ftruncate(fileno(out), write_pos) != 0)
if (fio_ftruncate(out, write_pos) != 0)
elog(ERROR, "Cannot truncate \"%s\": %s",
file->path, strerror(errno));
elog(VERBOSE, "Delta truncate file %s to block %u",
@ -904,21 +895,21 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
}
/* update file permission */
if (chmod(to_path, file->mode) == -1)
if (fio_chmod(to_path, file->mode, FIO_DB_HOST) == -1)
{
int errno_tmp = errno;
if (in)
fclose(in);
fclose(out);
fio_fclose(out);
elog(ERROR, "Cannot change mode of \"%s\": %s", to_path,
strerror(errno_tmp));
}
if (fflush(out) != 0 ||
fsync(fileno(out)) != 0 ||
fclose(out))
if (fio_fflush(out) != 0 ||
fio_fclose(out))
elog(ERROR, "Cannot write \"%s\": %s", to_path, strerror(errno));
if (in)
fclose(in);
}
@ -929,7 +920,8 @@ restore_data_file(const char *to_path, pgFile *file, bool allow_truncate,
* it is either small control file or already compressed cfs file.
*/
bool
copy_file(const char *from_root, const char *to_root, pgFile *file)
copy_file(const char *from_root, fio_location from_location,
const char *to_root, fio_location to_location, pgFile *file)
{
char to_path[MAXPGPATH];
FILE *in;
@ -947,7 +939,7 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
file->write_size = 0;
/* open backup mode file for read */
in = fopen(file->path, PG_BINARY_R);
in = fio_fopen(file->path, PG_BINARY_R, from_location);
if (in == NULL)
{
FIN_FILE_CRC32(true, crc);
@ -967,20 +959,20 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
/* open backup file for write */
join_path_components(to_path, to_root, file->path + strlen(from_root) + 1);
out = fopen(to_path, PG_BINARY_W);
out = fio_fopen(to_path, PG_BINARY_W, to_location);
if (out == NULL)
{
int errno_tmp = errno;
fclose(in);
fio_fclose(in);
elog(ERROR, "cannot open destination file \"%s\": %s",
to_path, strerror(errno_tmp));
}
/* stat source file to change mode of destination file */
if (fstat(fileno(in), &st) == -1)
if (fio_ffstat(in, &st) == -1)
{
fclose(in);
fclose(out);
fio_fclose(in);
fio_fclose(out);
elog(ERROR, "cannot stat \"%s\": %s", file->path,
strerror(errno));
}
@ -990,15 +982,15 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
{
read_len = 0;
if ((read_len = fread(buf, 1, sizeof(buf), in)) != sizeof(buf))
if ((read_len = fio_fread(in, buf, sizeof(buf))) != sizeof(buf))
break;
if (fwrite(buf, 1, read_len, out) != read_len)
if (fio_fwrite(out, buf, read_len) != read_len)
{
errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
fio_fclose(in);
fio_fclose(out);
elog(ERROR, "cannot write to \"%s\": %s", to_path,
strerror(errno_tmp));
}
@ -1009,10 +1001,10 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
}
errno_tmp = errno;
if (!feof(in))
if (read_len < 0)
{
fclose(in);
fclose(out);
fio_fclose(in);
fio_fclose(out);
elog(ERROR, "cannot read backup mode file \"%s\": %s",
file->path, strerror(errno_tmp));
}
@ -1020,12 +1012,12 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
/* copy odd part. */
if (read_len > 0)
{
if (fwrite(buf, 1, read_len, out) != read_len)
if (fio_fwrite(out, buf, read_len) != read_len)
{
errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
fio_fclose(in);
fio_fclose(out);
elog(ERROR, "cannot write to \"%s\": %s", to_path,
strerror(errno_tmp));
}
@ -1041,20 +1033,19 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
file->crc = crc;
/* update file permission */
if (chmod(to_path, st.st_mode) == -1)
if (fio_chmod(to_path, st.st_mode, to_location) == -1)
{
errno_tmp = errno;
fclose(in);
fclose(out);
fio_fclose(in);
fio_fclose(out);
elog(ERROR, "cannot change mode of \"%s\": %s", to_path,
strerror(errno_tmp));
}
if (fflush(out) != 0 ||
fsync(fileno(out)) != 0 ||
fclose(out))
if (fio_fflush(out) != 0 ||
fio_fclose(out))
elog(ERROR, "cannot write \"%s\": %s", to_path, strerror(errno));
fclose(in);
fio_fclose(in);
return true;
}
@ -1069,7 +1060,7 @@ get_gz_error(gzFile gzf, int errnum)
int gz_errnum;
const char *errmsg;
errmsg = gzerror(gzf, &gz_errnum);
errmsg = fio_gzerror(gzf, &gz_errnum);
if (gz_errnum == Z_ERRNO)
return strerror(errnum);
else
@ -1081,22 +1072,22 @@ get_gz_error(gzFile gzf, int errnum)
* Copy file attributes
*/
static void
copy_meta(const char *from_path, const char *to_path, bool unlink_on_error)
copy_meta(const char *from_path, fio_location from_location, const char *to_path, fio_location to_location, bool unlink_on_error)
{
struct stat st;
if (stat(from_path, &st) == -1)
if (fio_stat(from_path, &st, true, from_location) == -1)
{
if (unlink_on_error)
unlink(to_path);
fio_unlink(to_path, to_location);
elog(ERROR, "Cannot stat file \"%s\": %s",
from_path, strerror(errno));
}
if (chmod(to_path, st.st_mode) == -1)
if (fio_chmod(to_path, st.st_mode, to_location) == -1)
{
if (unlink_on_error)
unlink(to_path);
fio_unlink(to_path, to_location);
elog(ERROR, "Cannot change mode of file \"%s\": %s",
to_path, strerror(errno));
}
@ -1110,7 +1101,7 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
bool overwrite)
{
FILE *in = NULL;
int out;
int out = -1;
char buf[XLOG_BLCKSZ];
const char *to_path_p;
char to_path_temp[MAXPGPATH];
@ -1119,6 +1110,7 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
#ifdef HAVE_LIBZ
char gz_to_path[MAXPGPATH];
gzFile gz_out = NULL;
if (is_compress)
{
snprintf(gz_to_path, sizeof(gz_to_path), "%s.gz", to_path);
@ -1129,13 +1121,13 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
to_path_p = to_path;
/* open file for read */
in = fopen(from_path, PG_BINARY_R);
in = fio_fopen(from_path, PG_BINARY_R, FIO_DB_HOST);
if (in == NULL)
elog(ERROR, "Cannot open source WAL file \"%s\": %s", from_path,
strerror(errno));
/* Check if possible to skip copying */
if (fileExists(to_path_p))
if (fileExists(to_path_p, FIO_BACKUP_HOST))
{
if (fileEqualCRC(from_path, to_path_p, is_compress))
return;
@ -1150,25 +1142,17 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
{
snprintf(to_path_temp, sizeof(to_path_temp), "%s.partial", gz_to_path);
out = open(to_path_temp, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
S_IRUSR | S_IWUSR);
if (out < 0)
gz_out = fio_gzopen(to_path_temp, PG_BINARY_W, instance_config.compress_level, FIO_BACKUP_HOST);
if (gz_out == NULL)
elog(ERROR, "Cannot open destination temporary WAL file \"%s\": %s",
to_path_temp, strerror(errno));
gz_out = gzdopen(out, PG_BINARY_W);
if (gzsetparams(gz_out, instance_config.compress_level, Z_DEFAULT_STRATEGY) != Z_OK)
elog(ERROR, "Cannot set compression level %d to file \"%s\": %s",
instance_config.compress_level, to_path_temp,
get_gz_error(gz_out, errno));
}
else
#endif
{
snprintf(to_path_temp, sizeof(to_path_temp), "%s.partial", to_path);
out = open(to_path_temp, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
S_IRUSR | S_IWUSR);
out = fio_open(to_path_temp, O_RDWR | O_CREAT | O_EXCL | PG_BINARY, FIO_BACKUP_HOST);
if (out < 0)
elog(ERROR, "Cannot open destination temporary WAL file \"%s\": %s",
to_path_temp, strerror(errno));
@ -1177,14 +1161,14 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
/* copy content */
for (;;)
{
size_t read_len = 0;
ssize_t read_len = 0;
read_len = fread(buf, 1, sizeof(buf), in);
read_len = fio_fread(in, buf, sizeof(buf));
if (ferror(in))
if (read_len < 0)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_BACKUP_HOST);
elog(ERROR,
"Cannot read source WAL file \"%s\": %s",
from_path, strerror(errno_temp));
@ -1195,10 +1179,10 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
#ifdef HAVE_LIBZ
if (is_compress)
{
if (gzwrite(gz_out, buf, read_len) != read_len)
if (fio_gzwrite(gz_out, buf, read_len) != read_len)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot write to compressed WAL file \"%s\": %s",
to_path_temp, get_gz_error(gz_out, errno_temp));
}
@ -1206,27 +1190,27 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
else
#endif
{
if (write(out, buf, read_len) != read_len)
if (fio_write(out, buf, read_len) != read_len)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot write to WAL file \"%s\": %s",
to_path_temp, strerror(errno_temp));
}
}
}
if (feof(in) || read_len == 0)
if (read_len == 0)
break;
}
#ifdef HAVE_LIBZ
if (is_compress)
{
if (gzclose(gz_out) != 0)
if (fio_gzclose(gz_out) != 0)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot close compressed WAL file \"%s\": %s",
to_path_temp, get_gz_error(gz_out, errno_temp));
}
@ -1234,30 +1218,30 @@ push_wal_file(const char *from_path, const char *to_path, bool is_compress,
else
#endif
{
if (fsync(out) != 0 || close(out) != 0)
if (fio_flush(out) != 0 || fio_close(out) != 0)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot write WAL file \"%s\": %s",
to_path_temp, strerror(errno_temp));
}
}
if (fclose(in))
if (fio_fclose(in))
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot close source WAL file \"%s\": %s",
from_path, strerror(errno_temp));
}
/* update file permission. */
copy_meta(from_path, to_path_temp, true);
copy_meta(from_path, FIO_DB_HOST, to_path_temp, FIO_BACKUP_HOST, true);
if (rename(to_path_temp, to_path_p) < 0)
if (fio_rename(to_path_temp, to_path_p, FIO_BACKUP_HOST) < 0)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_BACKUP_HOST);
elog(ERROR, "Cannot rename WAL file \"%s\" to \"%s\": %s",
to_path_temp, to_path_p, strerror(errno_temp));
}
@ -1287,9 +1271,8 @@ get_wal_file(const char *from_path, const char *to_path)
gzFile gz_in = NULL;
#endif
/* open file for read */
in = fopen(from_path, PG_BINARY_R);
if (in == NULL)
/* First check source file for existance */
if (fio_access(from_path, F_OK, FIO_BACKUP_HOST) != 0)
{
#ifdef HAVE_LIBZ
/*
@ -1297,19 +1280,7 @@ get_wal_file(const char *from_path, const char *to_path)
* extension.
*/
snprintf(gz_from_path, sizeof(gz_from_path), "%s.gz", from_path);
gz_in = gzopen(gz_from_path, PG_BINARY_R);
if (gz_in == NULL)
{
if (errno == ENOENT)
{
/* There is no compressed file too, raise an error below */
}
/* Cannot open compressed file for some reason */
else
elog(ERROR, "Cannot open compressed WAL file \"%s\": %s",
gz_from_path, strerror(errno));
}
else
if (fio_access(gz_from_path, F_OK, FIO_BACKUP_HOST) == 0)
{
/* Found compressed file */
is_decompress = true;
@ -1318,15 +1289,33 @@ get_wal_file(const char *from_path, const char *to_path)
#endif
/* Didn't find compressed file */
if (!is_decompress)
elog(ERROR, "Cannot open source WAL file \"%s\": %s",
from_path, strerror(errno));
elog(ERROR, "Source WAL file \"%s\" doesn't exist",
from_path);
}
/* open file for read */
if (!is_decompress)
{
in = fio_fopen(from_path, PG_BINARY_R, FIO_BACKUP_HOST);
if (in == NULL)
elog(ERROR, "Cannot open source WAL file \"%s\": %s",
from_path, strerror(errno));
}
#ifdef HAVE_LIBZ
else
{
gz_in = fio_gzopen(gz_from_path, PG_BINARY_R, Z_DEFAULT_COMPRESSION,
FIO_BACKUP_HOST);
if (gz_in == NULL)
elog(ERROR, "Cannot open compressed WAL file \"%s\": %s",
gz_from_path, strerror(errno));
}
#endif
/* open backup file for write */
snprintf(to_path_temp, sizeof(to_path_temp), "%s.partial", to_path);
out = open(to_path_temp, O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
S_IRUSR | S_IWUSR);
out = fio_open(to_path_temp, O_RDWR | O_CREAT | O_EXCL | PG_BINARY, FIO_DB_HOST);
if (out < 0)
elog(ERROR, "Cannot open destination temporary WAL file \"%s\": %s",
to_path_temp, strerror(errno));
@ -1334,16 +1323,16 @@ get_wal_file(const char *from_path, const char *to_path)
/* copy content */
for (;;)
{
size_t read_len = 0;
int read_len = 0;
#ifdef HAVE_LIBZ
if (is_decompress)
{
read_len = gzread(gz_in, buf, sizeof(buf));
if (read_len != sizeof(buf) && !gzeof(gz_in))
read_len = fio_gzread(gz_in, buf, sizeof(buf));
if (read_len <= 0 && !fio_gzeof(gz_in))
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_DB_HOST);
elog(ERROR, "Cannot read compressed WAL file \"%s\": %s",
gz_from_path, get_gz_error(gz_in, errno_temp));
}
@ -1351,11 +1340,11 @@ get_wal_file(const char *from_path, const char *to_path)
else
#endif
{
read_len = fread(buf, 1, sizeof(buf), in);
read_len = fio_fread(in, buf, sizeof(buf));
if (ferror(in))
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_DB_HOST);
elog(ERROR, "Cannot read source WAL file \"%s\": %s",
from_path, strerror(errno_temp));
}
@ -1363,10 +1352,10 @@ get_wal_file(const char *from_path, const char *to_path)
if (read_len > 0)
{
if (write(out, buf, read_len) != read_len)
if (fio_write(out, buf, read_len) != read_len)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_DB_HOST);
elog(ERROR, "Cannot write to WAL file \"%s\": %s", to_path_temp,
strerror(errno_temp));
}
@ -1376,21 +1365,21 @@ get_wal_file(const char *from_path, const char *to_path)
#ifdef HAVE_LIBZ
if (is_decompress)
{
if (gzeof(gz_in) || read_len == 0)
if (fio_gzeof(gz_in) || read_len == 0)
break;
}
else
#endif
{
if (feof(in) || read_len == 0)
if (/* feof(in) || */ read_len == 0)
break;
}
}
if (fsync(out) != 0 || close(out) != 0)
if (fio_flush(out) != 0 || fio_close(out) != 0)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_DB_HOST);
elog(ERROR, "Cannot write WAL file \"%s\": %s",
to_path_temp, strerror(errno_temp));
}
@ -1398,10 +1387,10 @@ get_wal_file(const char *from_path, const char *to_path)
#ifdef HAVE_LIBZ
if (is_decompress)
{
if (gzclose(gz_in) != 0)
if (fio_gzclose(gz_in) != 0)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_DB_HOST);
elog(ERROR, "Cannot close compressed WAL file \"%s\": %s",
gz_from_path, get_gz_error(gz_in, errno_temp));
}
@ -1409,22 +1398,22 @@ get_wal_file(const char *from_path, const char *to_path)
else
#endif
{
if (fclose(in))
if (fio_fclose(in))
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_DB_HOST);
elog(ERROR, "Cannot close source WAL file \"%s\": %s",
from_path, strerror(errno_temp));
}
}
/* update file permission. */
copy_meta(from_path_p, to_path_temp, true);
copy_meta(from_path_p, FIO_BACKUP_HOST, to_path_temp, FIO_DB_HOST, true);
if (rename(to_path_temp, to_path) < 0)
if (fio_rename(to_path_temp, to_path, FIO_DB_HOST) < 0)
{
errno_temp = errno;
unlink(to_path_temp);
fio_unlink(to_path_temp, FIO_DB_HOST);
elog(ERROR, "Cannot rename WAL file \"%s\" to \"%s\": %s",
to_path_temp, to_path, strerror(errno_temp));
}
@ -1441,11 +1430,11 @@ get_wal_file(const char *from_path, const char *to_path)
* PG_TABLESPACE_MAP_FILE and PG_BACKUP_LABEL_FILE.
*/
void
calc_file_checksum(pgFile *file)
calc_file_checksum(pgFile *file, fio_location location)
{
Assert(S_ISREG(file->mode));
file->crc = pgFileGetCRC(file->path, true, false, &file->read_size);
file->crc = pgFileGetCRC(file->path, true, false, &file->read_size, location);
file->write_size = file->read_size;
}
@ -1724,7 +1713,7 @@ fileEqualCRC(const char *path1, const char *path2, bool path2_is_compressed)
gzFile gz_in = NULL;
INIT_FILE_CRC32(true, crc2);
gz_in = gzopen(path2, PG_BINARY_R);
gz_in = fio_gzopen(path2, PG_BINARY_R, Z_DEFAULT_COMPRESSION, FIO_BACKUP_HOST);
if (gz_in == NULL)
/* File cannot be read */
elog(ERROR,
@ -1733,32 +1722,33 @@ fileEqualCRC(const char *path1, const char *path2, bool path2_is_compressed)
for (;;)
{
size_t read_len = 0;
read_len = gzread(gz_in, buf, sizeof(buf));
if (read_len != sizeof(buf) && !gzeof(gz_in))
int read_len = fio_gzread(gz_in, buf, sizeof(buf));
if (read_len <= 0 && !fio_gzeof(gz_in))
{
/* An error occurred while reading the file */
elog(ERROR,
"Cannot compare WAL file \"%s\" with compressed \"%s\"",
path1, path2);
elog(WARNING,
"Cannot compare WAL file \"%s\" with compressed \"%s\": %d",
path1, path2, read_len);
return false;
}
COMP_FILE_CRC32(true, crc2, buf, read_len);
if (gzeof(gz_in) || read_len == 0)
if (fio_gzeof(gz_in) || read_len == 0)
break;
}
FIN_FILE_CRC32(true, crc2);
if (gzclose(gz_in) != 0)
if (fio_gzclose(gz_in) != 0)
elog(ERROR, "Cannot close compressed WAL file \"%s\": %s",
path2, get_gz_error(gz_in, errno));
}
else
#endif
{
crc2 = pgFileGetCRC(path2, true, true, NULL);
crc2 = pgFileGetCRC(path2, true, true, NULL, FIO_BACKUP_HOST);
}
/* Get checksum of original file */
crc1 = pgFileGetCRC(path1, true, true, NULL);
crc1 = pgFileGetCRC(path1, true, true, NULL, FIO_DB_HOST);
return EQ_CRC32C(crc1, crc2);
}

4
src/delete.c
View File

@ -658,7 +658,7 @@ delete_backup_files(pgBackup *backup)
/* list files to be deleted */
files = parray_new();
pgBackupGetPath(backup, path, lengthof(path), NULL);
dir_list_file(files, path, false, true, true, 0);
dir_list_file(files, path, false, true, true, 0, FIO_BACKUP_HOST);
/* delete leaf node first */
parray_qsort(files, pgFileComparePathDesc);
@ -674,7 +674,7 @@ delete_backup_files(pgBackup *backup)
if (interrupted)
elog(ERROR, "interrupted during delete backup");
pgFileDelete(file);
fio_unlink(file->path, FIO_BACKUP_HOST);
}
parray_walk(files, pgFileFree);

103
src/dir.c
View File

@ -9,6 +9,8 @@
*/
#include "pg_probackup.h"
#include "utils/file.h"
#if PG_VERSION_NUM < 110000
#include "catalog/catalog.h"
@ -121,11 +123,10 @@ static int BlackListCompare(const void *str1, const void *str2);
static char dir_check_file(const char *root, pgFile *file);
static void dir_list_file_internal(parray *files, const char *root,
pgFile *parent, bool exclude,
bool omit_symlink, parray *black_list,
int external_dir_num);
bool omit_symlink, parray *black_list, int external_dir_num, fio_location location);
static void list_data_directories(parray *files, const char *path, bool is_root,
bool exclude);
bool exclude, fio_location location);
static void opt_path_map(ConfigOption *opt, const char *arg,
TablespaceList *list, const char *type);
@ -162,13 +163,13 @@ dir_create_dir(const char *dir, mode_t mode)
}
pgFile *
pgFileNew(const char *path, bool omit_symlink, int external_dir_num)
pgFileNew(const char *path, bool omit_symlink, int external_dir_num, fio_location location)
{
struct stat st;
pgFile *file;
/* stat the file */
if ((omit_symlink ? stat(path, &st) : lstat(path, &st)) == -1)
/* stat the file */
if (fio_stat(path, &st, omit_symlink, location) < 0)
{
/* file not found is not an error case */
if (errno == ENOENT)
@ -269,19 +270,19 @@ delete_file:
pg_crc32
pgFileGetCRC(const char *file_path, bool use_crc32c, bool raise_on_deleted,
size_t *bytes_read)
size_t *bytes_read, fio_location location)
{
FILE *fp;
pg_crc32 crc = 0;
char buf[1024];
size_t len;
size_t len = 0;
size_t total = 0;
int errno_tmp;
INIT_FILE_CRC32(use_crc32c, crc);
/* open file in binary read mode */
fp = fopen(file_path, PG_BINARY_R);
fp = fio_fopen(file_path, PG_BINARY_R, location);
if (fp == NULL)
{
if (!raise_on_deleted && errno == ENOENT)
@ -300,7 +301,7 @@ pgFileGetCRC(const char *file_path, bool use_crc32c, bool raise_on_deleted,
if (interrupted)
elog(ERROR, "interrupted during CRC calculation");
len = fread(buf, 1, sizeof(buf), fp);
len = fio_fread(fp, buf, sizeof(buf));
if(len == 0)
break;
/* update CRC */
@ -312,12 +313,12 @@ pgFileGetCRC(const char *file_path, bool use_crc32c, bool raise_on_deleted,
*bytes_read = total;
errno_tmp = errno;
if (!feof(fp))
if (len < 0)
elog(WARNING, "cannot read \"%s\": %s", file_path,
strerror(errno_tmp));
FIN_FILE_CRC32(use_crc32c, crc);
fclose(fp);
fio_fclose(fp);
return crc;
}
@ -433,7 +434,7 @@ BlackListCompare(const void *str1, const void *str2)
*/
void
dir_list_file(parray *files, const char *root, bool exclude, bool omit_symlink,
bool add_root, int external_dir_num)
bool add_root, int external_dir_num, fio_location location)
{
pgFile *file;
parray *black_list = NULL;
@ -442,14 +443,14 @@ dir_list_file(parray *files, const char *root, bool exclude, bool omit_symlink,
join_path_components(path, backup_instance_path, PG_BLACK_LIST);
/* List files with black list */
if (root && instance_config.pgdata &&
strcmp(root, instance_config.pgdata) == 0 && fileExists(path))
strcmp(root, instance_config.pgdata) == 0 && fileExists(path, FIO_BACKUP_HOST))
{
FILE *black_list_file = NULL;
char buf[MAXPGPATH * 2];
char black_item[MAXPGPATH * 2];
black_list = parray_new();
black_list_file = fopen(path, PG_BINARY_R);
black_list_file = fio_open_stream(path, FIO_BACKUP_HOST);
if (black_list_file == NULL)
elog(ERROR, "cannot open black_list: %s", strerror(errno));
@ -468,11 +469,11 @@ dir_list_file(parray *files, const char *root, bool exclude, bool omit_symlink,
parray_append(black_list, pgut_strdup(black_item));
}
fclose(black_list_file);
fio_close_stream(black_list_file);
parray_qsort(black_list, BlackListCompare);
}
file = pgFileNew(root, omit_symlink, external_dir_num);
file = pgFileNew(root, omit_symlink, external_dir_num, location);
if (file == NULL)
return;
@ -489,7 +490,7 @@ dir_list_file(parray *files, const char *root, bool exclude, bool omit_symlink,
parray_append(files, file);
dir_list_file_internal(files, root, file, exclude, omit_symlink, black_list,
external_dir_num);
external_dir_num, location);
if (!add_root)
pgFileFree(file);
@ -709,7 +710,7 @@ dir_check_file(const char *root, pgFile *file)
static void
dir_list_file_internal(parray *files, const char *root, pgFile *parent,
bool exclude, bool omit_symlink, parray *black_list,
int external_dir_num)
int external_dir_num, fio_location location)
{
DIR *dir;
struct dirent *dent;
@ -718,7 +719,7 @@ dir_list_file_internal(parray *files, const char *root, pgFile *parent,
elog(ERROR, "\"%s\" is not a directory", parent->path);
/* Open directory and list contents */
dir = opendir(parent->path);
dir = fio_opendir(parent->path, location);
if (dir == NULL)
{
if (errno == ENOENT)
@ -731,7 +732,7 @@ dir_list_file_internal(parray *files, const char *root, pgFile *parent,
}
errno = 0;
while ((dent = readdir(dir)))
while ((dent = fio_readdir(dir)))
{
pgFile *file;
char child[MAXPGPATH];
@ -739,7 +740,7 @@ dir_list_file_internal(parray *files, const char *root, pgFile *parent,
join_path_components(child, parent->path, dent->d_name);
file = pgFileNew(child, omit_symlink, external_dir_num);
file = pgFileNew(child, omit_symlink, external_dir_num, location);
if (file == NULL)
continue;
@ -796,17 +797,17 @@ dir_list_file_internal(parray *files, const char *root, pgFile *parent,
*/
if (S_ISDIR(file->mode))
dir_list_file_internal(files, root, file, exclude, omit_symlink,
black_list, external_dir_num);
black_list, external_dir_num, location);
}
if (errno && errno != ENOENT)
{
int errno_tmp = errno;
closedir(dir);
fio_closedir(dir);
elog(ERROR, "cannot read directory \"%s\": %s",
parent->path, strerror(errno_tmp));
}
closedir(dir);
fio_closedir(dir);
}
/*
@ -818,7 +819,7 @@ dir_list_file_internal(parray *files, const char *root, pgFile *parent,
*/
static void
list_data_directories(parray *files, const char *path, bool is_root,
bool exclude)
bool exclude, fio_location location)
{
DIR *dir;
struct dirent *dent;
@ -826,12 +827,12 @@ list_data_directories(parray *files, const char *path, bool is_root,
bool has_child_dirs = false;
/* open directory and list contents */
dir = opendir(path);
dir = fio_opendir(path, location);
if (dir == NULL)
elog(ERROR, "cannot open directory \"%s\": %s", path, strerror(errno));
errno = 0;
while ((dent = readdir(dir)))
while ((dent = fio_readdir(dir)))
{
char child[MAXPGPATH];
bool skip = false;
@ -844,7 +845,7 @@ list_data_directories(parray *files, const char *path, bool is_root,
join_path_components(child, path, dent->d_name);
if (lstat(child, &st) == -1)
if (fio_stat(child, &st, false, location) == -1)
elog(ERROR, "cannot stat file \"%s\": %s", child, strerror(errno));
if (!S_ISDIR(st.st_mode))
@ -868,7 +869,7 @@ list_data_directories(parray *files, const char *path, bool is_root,
continue;
has_child_dirs = true;
list_data_directories(files, child, false, exclude);
list_data_directories(files, child, false, exclude, location);
}
/* List only full and last directories */
@ -876,12 +877,12 @@ list_data_directories(parray *files, const char *path, bool is_root,
{
pgFile *dir;
dir = pgFileNew(path, false, 0);
dir = pgFileNew(path, false, 0, location);
parray_append(files, dir);
}
prev_errno = errno;
closedir(dir);
fio_closedir(dir);
if (prev_errno && prev_errno != ENOENT)
elog(ERROR, "cannot read directory \"%s\": %s",
@ -1025,14 +1026,13 @@ opt_externaldir_map(ConfigOption *opt, const char *arg)
*/
void
create_data_directories(const char *data_dir, const char *backup_dir,
bool extract_tablespaces)
bool extract_tablespaces, fio_location location)
{
parray *dirs,
*links = NULL;
size_t i;
char backup_database_dir[MAXPGPATH],
to_path[MAXPGPATH];
dirs = parray_new();
if (extract_tablespaces)
{
@ -1043,7 +1043,8 @@ create_data_directories(const char *data_dir, const char *backup_dir,
}
join_path_components(backup_database_dir, backup_dir, DATABASE_DIR);
list_data_directories(dirs, backup_database_dir, true, false);
list_data_directories(dirs, backup_database_dir, true, false,
FIO_BACKUP_HOST);
elog(LOG, "restore directories and symlinks...");
@ -1125,15 +1126,15 @@ create_data_directories(const char *data_dir, const char *backup_dir,
linked_path, relative_ptr);
/* Firstly, create linked directory */
dir_create_dir(linked_path, DIR_PERMISSION);
fio_mkdir(linked_path, DIR_PERMISSION, location);
join_path_components(to_path, data_dir, PG_TBLSPC_DIR);
/* Create pg_tblspc directory just in case */
dir_create_dir(to_path, DIR_PERMISSION);
fio_mkdir(to_path, DIR_PERMISSION, location);
/* Secondly, create link */
join_path_components(to_path, to_path, link_name);
if (symlink(linked_path, to_path) < 0)
if (fio_symlink(linked_path, to_path, location) < 0)
elog(ERROR, "could not create symbolic link \"%s\": %s",
to_path, strerror(errno));
@ -1156,7 +1157,7 @@ create_directory:
/* This is not symlink, create directory */
join_path_components(to_path, data_dir, relative_ptr);
dir_create_dir(to_path, DIR_PERMISSION);
fio_mkdir(to_path, DIR_PERMISSION, location);
}
if (extract_tablespaces)
@ -1185,13 +1186,13 @@ read_tablespace_map(parray *files, const char *backup_dir)
join_path_components(map_path, db_path, PG_TABLESPACE_MAP_FILE);
/* Exit if database/tablespace_map doesn't exist */
if (!fileExists(map_path))
if (!fileExists(map_path, FIO_BACKUP_HOST))
{
elog(LOG, "there is no file tablespace_map");
return;
}
fp = fopen(map_path, "rt");
fp = fio_open_stream(map_path, FIO_BACKUP_HOST);
if (fp == NULL)
elog(ERROR, "cannot open \"%s\": %s", map_path, strerror(errno));
@ -1216,7 +1217,7 @@ read_tablespace_map(parray *files, const char *backup_dir)
parray_append(files, file);
}
fclose(fp);
fio_close_stream(fp);
}
/*
@ -1365,7 +1366,7 @@ print_file_list(FILE *out, const parray *files, const char *root,
file->external_dir_num - 1));
}
fprintf(out, "{\"path\":\"%s\", \"size\":\"" INT64_FORMAT "\", "
fio_fprintf(out, "{\"path\":\"%s\", \"size\":\"" INT64_FORMAT "\", "
"\"mode\":\"%u\", \"is_datafile\":\"%u\", "
"\"is_cfs\":\"%u\", \"crc\":\"%u\", "
"\"compress_alg\":\"%s\", \"external_dir_num\":\"%d\"",
@ -1374,15 +1375,15 @@ print_file_list(FILE *out, const parray *files, const char *root,
deparse_compress_alg(file->compress_alg), file->external_dir_num);
if (file->is_datafile)
fprintf(out, ",\"segno\":\"%d\"", file->segno);
fio_fprintf(out, ",\"segno\":\"%d\"", file->segno);
if (file->linked)
fprintf(out, ",\"linked\":\"%s\"", file->linked);
if (file->n_blocks != BLOCKNUM_INVALID)
fprintf(out, ",\"n_blocks\":\"%i\"", file->n_blocks);
fio_fprintf(out, ",\"n_blocks\":\"%i\"", file->n_blocks);
fprintf(out, "}\n");
fio_fprintf(out, "}\n");
}
}
@ -1535,13 +1536,13 @@ bad_format:
*/
parray *
dir_read_file_list(const char *root, const char *external_prefix,
const char *file_txt)
const char *file_txt, fio_location location)
{
FILE *fp;
parray *files;
char buf[MAXPGPATH * 2];
fp = fopen(file_txt, "rt");
fp = fio_open_stream(file_txt, location);
if (fp == NULL)
elog(ERROR, "cannot open \"%s\": %s", file_txt, strerror(errno));
@ -1610,7 +1611,7 @@ dir_read_file_list(const char *root, const char *external_prefix,
parray_append(files, file);
}
fclose(fp);
fio_close_stream(fp);
return files;
}
@ -1656,11 +1657,11 @@ dir_is_empty(const char *path)
* Return true if the path is a existing regular file.
*/
bool
fileExists(const char *path)
fileExists(const char *path, fio_location location)
{
struct stat buf;
if (stat(path, &buf) == -1 && errno == ENOENT)
if (fio_stat(path, &buf, true, location) == -1 && errno == ENOENT)
return false;
else if (!S_ISREG(buf.st_mode))
return false;

20
src/fetch.c
View File

@ -25,7 +25,7 @@
*
*/
char *
slurpFile(const char *datadir, const char *path, size_t *filesize, bool safe)
slurpFile(const char *datadir, const char *path, size_t *filesize, bool safe, fio_location location)
{
int fd;
char *buffer;
@ -34,7 +34,16 @@ slurpFile(const char *datadir, const char *path, size_t *filesize, bool safe)
int len;
snprintf(fullpath, sizeof(fullpath), "%s/%s", datadir, path);
if ((fd = open(fullpath, O_RDONLY | PG_BINARY, 0)) == -1)
if (fio_access(fullpath, R_OK, location) != 0)
{
if (safe)
return NULL;
else
elog(ERROR, "could not open file \"%s\" for reading: %s",
fullpath, strerror(errno));
}
if ((fd = fio_open(fullpath, O_RDONLY | PG_BINARY, location)) == -1)
{
if (safe)
return NULL;
@ -43,7 +52,7 @@ slurpFile(const char *datadir, const char *path, size_t *filesize, bool safe)
fullpath, strerror(errno));
}
if (fstat(fd, &statbuf) < 0)
if (fio_fstat(fd, &statbuf) < 0)
{
if (safe)
return NULL;
@ -53,10 +62,9 @@ slurpFile(const char *datadir, const char *path, size_t *filesize, bool safe)
}
len = statbuf.st_size;
buffer = pg_malloc(len + 1);
if (read(fd, buffer, len) != len)
if (fio_read(fd, buffer, len) != len)
{
if (safe)
return NULL;
@ -65,7 +73,7 @@ slurpFile(const char *datadir, const char *path, size_t *filesize, bool safe)
fullpath, strerror(errno));
}
close(fd);
fio_close(fd);
/* Zero-terminate the buffer. */
buffer[len] = '\0';

72
src/help.c
View File

@ -120,6 +120,9 @@ help_pg_probackup(void)
printf(_(" [--replica-timeout=timeout]\n"));
printf(_(" [--no-validate] [--skip-block-validation]\n"));
printf(_(" [--external-dirs=external-directory-path]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
printf(_("\n %s restore -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" [-D pgdata-path] [-i backup-id] [-j num-threads]\n"));
@ -133,6 +136,9 @@ help_pg_probackup(void)
printf(_(" [-T OLDDIR=NEWDIR] [--progress]\n"));
printf(_(" [--external-mapping=OLDDIR=NEWDIR]\n"));
printf(_(" [--skip-external-dirs]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
printf(_("\n %s validate -B backup-path [--instance=instance_name]\n"), PROGRAM_NAME);
printf(_(" [-i backup-id] [--progress] [-j num-threads]\n"));
@ -154,6 +160,9 @@ help_pg_probackup(void)
printf(_("\n %s add-instance -B backup-path -D pgdata-path\n"), PROGRAM_NAME);
printf(_(" --instance=instance_name\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
printf(_("\n %s del-instance -B backup-path\n"), PROGRAM_NAME);
printf(_(" --instance=instance_name\n"));
@ -165,10 +174,16 @@ help_pg_probackup(void)
printf(_(" [--compress-algorithm=compress-algorithm]\n"));
printf(_(" [--compress-level=compress-level]\n"));
printf(_(" [--overwrite]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
printf(_("\n %s archive-get -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" --wal-file-path=wal-file-path\n"));
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
if ((PROGRAM_URL || PROGRAM_EMAIL))
{
@ -214,7 +229,10 @@ help_backup(void)
printf(_(" [--master-port=port] [--master-user=user_name]\n"));
printf(_(" [--replica-timeout=timeout]\n"));
printf(_(" [--no-validate] [--skip-block-validation]\n"));
printf(_(" [-E external-dirs=external-directory-path]\n\n"));
printf(_(" [--external-dirs=external-directory-path]\n\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n"));
printf(_(" -B, --backup-path=backup-path location of the backup storage area\n"));
printf(_(" -b, --backup-mode=backup-mode backup mode=FULL|PAGE|DELTA|PTRACK\n"));
@ -287,6 +305,15 @@ help_backup(void)
printf(_(" --master-host=host_name database server host of master (deprecated)\n"));
printf(_(" --master-port=port database server port of master (deprecated)\n"));
printf(_(" --replica-timeout=timeout wait timeout for WAL segment streaming through replication (deprecated)\n"));
printf(_("\n Remote options:\n"));
printf(_(" --remote-proto=protocol remote protocol to use\n"));
printf(_(" available options: 'ssh', 'none' (default: none)\n"));
printf(_(" --remote-host=hostname remote host address or hostname\n"));
printf(_(" --remote-port=port remote host port (default: 22)\n"));
printf(_(" --remote-path=path path to pg_probackup binary on remote host (default: current binary path)\n"));
printf(_(" --remote-user=username user name for ssh connection (default current user)\n"));
printf(_(" --ssh-options=ssh_options additional ssh options (default: none)\n"));
}
static void
@ -304,6 +331,9 @@ help_restore(void)
printf(_(" [-T OLDDIR=NEWDIR] [--progress]\n"));
printf(_(" [--external-mapping=OLDDIR=NEWDIR]\n"));
printf(_(" [--skip-external-dirs]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n\n"));
printf(_(" -B, --backup-path=backup-path location of the backup storage area\n"));
printf(_(" --instance=instance_name name of the instance\n"));
@ -359,6 +389,15 @@ help_restore(void)
printf(_(" --log-rotation-age=log-rotation-age\n"));
printf(_(" rotate logfile if its age exceeds this value; 0 disables; (default: 0)\n"));
printf(_(" available units: 'ms', 's', 'min', 'h', 'd' (default: min)\n"));
printf(_("\n Remote options:\n"));
printf(_(" --remote-proto=protocol remote protocol to use\n"));
printf(_(" available options: 'ssh', 'none' (default: none)\n"));
printf(_(" --remote-host=hostname remote host address or hostname\n"));
printf(_(" --remote-port=port remote host port (default: 22)\n"));
printf(_(" --remote-path=path path to pg_probackup binary on remote host (default: current binary path)\n"));
printf(_(" --remote-user=username user name for ssh connection (default current user)\n"));
printf(_(" --ssh-options=ssh_options additional ssh options (default: none)\n"));
}
static void
@ -600,14 +639,26 @@ help_add_instance(void)
{
printf(_("%s add-instance -B backup-path -D pgdata-path\n"), PROGRAM_NAME);
printf(_(" --instance=instance_name\n"));
printf(_(" --remote-proto --remote-host\n"));
printf(_(" --remote-port --remote-path --remote-user\n"));
printf(_(" --ssh-options\n"));
printf(_(" -E external-dirs=external-directory-path\n\n"));
printf(_(" -B, --backup-path=backup-path location of the backup storage area\n"));
printf(_(" -D, --pgdata=pgdata-path location of the database storage area\n"));
printf(_(" --instance=instance_name name of the new instance\n"));
printf(_(" -E --external-dirs=external-directory-path\n"));
printf(_(" backup some directories not from pgdata \n"));
printf(_(" (example: --external-dirs=/tmp/dir1:/tmp/dir2)\n"));
printf(_("\n Remote options:\n"));
printf(_(" --remote-proto=protocol remote protocol to use\n"));
printf(_(" available options: 'ssh', 'none' (default: none)\n"));
printf(_(" --remote-host=hostname remote host address or hostname\n"));
printf(_(" --remote-port=port remote host port (default: 22)\n"));
printf(_(" --remote-path=path path to pg_probackup binary on remote host (default: current binary path)\n"));
printf(_(" --remote-user=username user name for ssh connection (default current user)\n"));
printf(_(" --ssh-options=ssh_options additional ssh options (default: none)\n"));
}
static void
@ -628,7 +679,10 @@ help_archive_push(void)
printf(_(" [--compress]\n"));
printf(_(" [--compress-algorithm=compress-algorithm]\n"));
printf(_(" [--compress-level=compress-level]\n"));
printf(_(" [--overwrite]\n\n"));
printf(_(" [--overwrite]\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n\n"));
printf(_(" -B, --backup-path=backup-path location of the backup storage area\n"));
printf(_(" --instance=instance_name name of the instance to delete\n"));
@ -649,7 +703,10 @@ help_archive_get(void)
{
printf(_("\n %s archive-get -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
printf(_(" --wal-file-path=wal-file-path\n"));
printf(_(" --wal-file-name=wal-file-name\n\n"));
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" [--remote-proto] [--remote-host]\n"));
printf(_(" [--remote-port] [--remote-path] [--remote-user]\n"));
printf(_(" [--ssh-options]\n\n"));
printf(_(" -B, --backup-path=backup-path location of the backup storage area\n"));
printf(_(" --instance=instance_name name of the instance to delete\n"));
@ -657,4 +714,13 @@ help_archive_get(void)
printf(_(" relative destination path name of the WAL file on the server\n"));
printf(_(" --wal-file-name=wal-file-name\n"));
printf(_(" name of the WAL file to retrieve from the archive\n"));
printf(_("\n Remote options:\n"));
printf(_(" --remote-proto=protocol remote protocol to use\n"));
printf(_(" available options: 'ssh', 'none' (default: none)\n"));
printf(_(" --remote-host=hostname remote host address or hostname\n"));
printf(_(" --remote-port=port remote host port (default: 22)\n"));
printf(_(" --remote-path=path path to pg_probackup binary on remote host (default: current binary path)\n"));
printf(_(" --remote-user=username user name for ssh connection (default current user)\n"));
printf(_(" --ssh-options=ssh_options additional ssh options (default: none)\n"));
}

18
src/merge.c
View File

@ -233,7 +233,7 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup)
*/
pgBackupGetPath(to_backup, control_file, lengthof(control_file),
DATABASE_FILE_LIST);
to_files = dir_read_file_list(NULL, NULL, control_file);
to_files = dir_read_file_list(NULL, NULL, control_file, FIO_BACKUP_HOST);
/* To delete from leaf, sort in reversed order */
parray_qsort(to_files, pgFileComparePathDesc);
/*
@ -241,7 +241,7 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup)
*/
pgBackupGetPath(from_backup, control_file, lengthof(control_file),
DATABASE_FILE_LIST);
files = dir_read_file_list(NULL, NULL, control_file);
files = dir_read_file_list(NULL, NULL, control_file, FIO_BACKUP_HOST);
/* sort by size for load balancing */
parray_qsort(files, pgFileCompareSize);
@ -255,7 +255,7 @@ merge_backups(pgBackup *to_backup, pgBackup *from_backup)
write_backup_status(to_backup, BACKUP_STATUS_MERGING);
write_backup_status(from_backup, BACKUP_STATUS_MERGING);
create_data_directories(to_database_path, from_backup_path, false);
create_data_directories(to_database_path, from_backup_path, false, FIO_BACKUP_HOST);
threads = (pthread_t *) palloc(sizeof(pthread_t) * num_threads);
threads_args = (merge_files_arg *) palloc(sizeof(merge_files_arg) * num_threads);
@ -505,7 +505,7 @@ merge_files(void *arg)
* Recalculate crc for backup prior to 2.0.25.
*/
if (parse_program_version(from_backup->program_version) < 20025)
file->crc = pgFileGetCRC(to_file_path, true, true, NULL);
file->crc = pgFileGetCRC(to_file_path, true, true, NULL, FIO_LOCAL_HOST);
/* Otherwise just get it from the previous file */
else
file->crc = to_file->crc;
@ -628,7 +628,7 @@ merge_files(void *arg)
* do that.
*/
file->write_size = pgFileSize(to_file_path);
file->crc = pgFileGetCRC(to_file_path, true, true, NULL);
file->crc = pgFileGetCRC(to_file_path, true, true, NULL, FIO_LOCAL_HOST);
}
}
else if (file->external_dir_num)
@ -646,12 +646,12 @@ merge_files(void *arg)
file->external_dir_num);
makeExternalDirPathByNum(to_root, argument->to_external_prefix,
new_dir_num);
copy_file(from_root, to_root, file);
copy_file(from_root, FIO_LOCAL_HOST, to_root, FIO_LOCAL_HOST, file);
}
else if (strcmp(file->name, "pg_control") == 0)
copy_pgcontrol_file(argument->from_root, argument->to_root, file);
copy_pgcontrol_file(argument->from_root, FIO_LOCAL_HOST, argument->to_root, FIO_LOCAL_HOST, file);
else
copy_file(argument->from_root, argument->to_root, file);
copy_file(argument->from_root, FIO_LOCAL_HOST, argument->to_root, FIO_LOCAL_HOST, file);
/*
* We need to save compression algorithm type of the target backup to be
@ -680,7 +680,7 @@ remove_dir_with_files(const char *path)
parray *files = parray_new();
int i;
dir_list_file(files, path, true, true, true, 0);
dir_list_file(files, path, true, true, true, 0, FIO_LOCAL_HOST);
parray_qsort(files, pgFileComparePathDesc);
for (i = 0; i < parray_num(files); i++)
{

34
src/parsexlog.c
View File

@ -103,8 +103,8 @@ typedef struct XLogReaderData
XLogSegNo xlogsegno;
bool xlogexists;
char page_buf[XLOG_BLCKSZ];
uint32 prev_page_off;
char page_buf[XLOG_BLCKSZ];
uint32 prev_page_off;
bool need_switch;
@ -112,8 +112,8 @@ typedef struct XLogReaderData
char xlogpath[MAXPGPATH];
#ifdef HAVE_LIBZ
gzFile gz_xlogfile;
char gz_xlogpath[MAXPGPATH];
gzFile gz_xlogfile;
char gz_xlogpath[MAXPGPATH];
#endif
} XLogReaderData;
@ -640,7 +640,7 @@ get_gz_error(gzFile gzf)
int errnum;
const char *errmsg;
errmsg = gzerror(gzf, &errnum);
errmsg = fio_gzerror(gzf, &errnum);
if (errnum == Z_ERRNO)
return strerror(errno);
else
@ -718,14 +718,14 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
snprintf(reader_data->xlogpath, MAXPGPATH, "%s/%s", wal_archivedir,
xlogfname);
if (fileExists(reader_data->xlogpath))
if (fileExists(reader_data->xlogpath, FIO_BACKUP_HOST))
{
elog(LOG, "Thread [%d]: Opening WAL segment \"%s\"",
reader_data->thread_num, reader_data->xlogpath);
reader_data->xlogexists = true;
reader_data->xlogfile = open(reader_data->xlogpath,
O_RDONLY | PG_BINARY, 0);
reader_data->xlogfile = fio_open(reader_data->xlogpath,
O_RDONLY | PG_BINARY, FIO_BACKUP_HOST);
if (reader_data->xlogfile < 0)
{
@ -741,14 +741,14 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
{
snprintf(reader_data->gz_xlogpath, sizeof(reader_data->gz_xlogpath),
"%s.gz", reader_data->xlogpath);
if (fileExists(reader_data->gz_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);
reader_data->xlogexists = true;
reader_data->gz_xlogfile = gzopen(reader_data->gz_xlogpath,
"rb");
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",
@ -784,14 +784,14 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
/* Read the requested page */
if (reader_data->xlogfile != -1)
{
if (lseek(reader_data->xlogfile, (off_t) targetPageOff, SEEK_SET) < 0)
if (fio_seek(reader_data->xlogfile, (off_t) targetPageOff) < 0)
{
elog(WARNING, "Thread [%d]: Could not seek in WAL segment \"%s\": %s",
reader_data->thread_num, reader_data->xlogpath, strerror(errno));
return -1;
}
if (read(reader_data->xlogfile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
if (fio_read(reader_data->xlogfile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
{
elog(WARNING, "Thread [%d]: Could not read from WAL segment \"%s\": %s",
reader_data->thread_num, reader_data->xlogpath, strerror(errno));
@ -801,7 +801,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
#ifdef HAVE_LIBZ
else
{
if (gzseek(reader_data->gz_xlogfile, (z_off_t) targetPageOff, SEEK_SET) == -1)
if (fio_gzseek(reader_data->gz_xlogfile, (z_off_t) targetPageOff, SEEK_SET) == -1)
{
elog(WARNING, "Thread [%d]: Could not seek in compressed WAL segment \"%s\": %s",
reader_data->thread_num, reader_data->gz_xlogpath,
@ -809,7 +809,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
return -1;
}
if (gzread(reader_data->gz_xlogfile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
if (fio_gzread(reader_data->gz_xlogfile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
{
elog(WARNING, "Thread [%d]: Could not read from compressed WAL segment \"%s\": %s",
reader_data->thread_num, reader_data->gz_xlogpath,
@ -1350,13 +1350,13 @@ CleanupXLogPageRead(XLogReaderState *xlogreader)
reader_data = (XLogReaderData *) xlogreader->private_data;
if (reader_data->xlogfile >= 0)
{
close(reader_data->xlogfile);
fio_close(reader_data->xlogfile);
reader_data->xlogfile = -1;
}
#ifdef HAVE_LIBZ
else if (reader_data->gz_xlogfile != NULL)
{
gzclose(reader_data->gz_xlogfile);
fio_gzclose(reader_data->gz_xlogfile);
reader_data->gz_xlogfile = NULL;
}
#endif

59
src/pg_probackup.c
View File

@ -12,6 +12,7 @@
#include "pg_getopt.h"
#include "streamutil.h"
#include "utils/file.h"
#include <sys/stat.h>
@ -19,7 +20,6 @@
#include "utils/thread.h"
#include <time.h>
const char *PROGRAM_VERSION = "2.0.27";
const char *PROGRAM_URL = "https://github.com/postgrespro/pg_probackup";
const char *PROGRAM_EMAIL = "https://github.com/postgrespro/pg_probackup/issues";
const char *PROGRAM_FULL_PATH = NULL;
@ -42,6 +42,9 @@ typedef enum ProbackupSubcmd
SHOW_CONFIG_CMD
} ProbackupSubcmd;
char *pg_probackup; /* Program name (argv[0]) */
/* directory options */
char *backup_path = NULL;
/*
@ -70,7 +73,7 @@ bool temp_slot = false;
/* backup options */
bool backup_logs = false;
bool smooth_checkpoint;
bool is_remote_backup = false;
char *remote_agent;
/* restore options */
static char *target_time = NULL;
@ -146,8 +149,6 @@ static ConfigOption cmd_options[] =
{ 'b', 135, "delete-expired", &delete_expired, SOURCE_CMD_STRICT },
{ 'b', 235, "merge-expired", &merge_expired, SOURCE_CMD_STRICT },
{ 'b', 237, "dry-run", &dry_run, SOURCE_CMD_STRICT },
/* TODO not completed feature. Make it unavailiable from user level
{ 'b', 18, "remote", &is_remote_backup, SOURCE_CMD_STRICT, }, */
/* restore options */
{ 's', 136, "recovery-target-time", &target_time, SOURCE_CMD_STRICT },
{ 's', 137, "recovery-target-xid", &target_xid, SOURCE_CMD_STRICT },
@ -193,6 +194,18 @@ static ConfigOption cmd_options[] =
{ 0 }
};
static void
setMyLocation(void)
{
MyLocation = IsSshProtocol()
? (backup_subcmd == ARCHIVE_PUSH_CMD || backup_subcmd == ARCHIVE_GET_CMD)
? FIO_DB_HOST
: (backup_subcmd == BACKUP_CMD || backup_subcmd == RESTORE_CMD)
? FIO_BACKUP_HOST
: FIO_LOCAL_HOST
: FIO_LOCAL_HOST;
}
/*
* Entry point of pg_probackup command.
*/
@ -205,6 +218,8 @@ main(int argc, char *argv[])
struct stat stat_buf;
int rc;
pg_probackup = argv[0];
/* Initialize current backup */
pgBackupInit(&current);
@ -264,6 +279,19 @@ main(int argc, char *argv[])
backup_subcmd = SET_CONFIG_CMD;
else if (strcmp(argv[1], "show-config") == 0)
backup_subcmd = SHOW_CONFIG_CMD;
else if (strcmp(argv[1], "agent") == 0 && argc > 2)
{
remote_agent = argv[2];
if (strcmp(remote_agent, PROGRAM_VERSION) != 0)
{
uint32 agent_version = parse_program_version(remote_agent);
elog(agent_version < AGENT_PROTOCOL_VERSION ? ERROR : WARNING,
"Agent version %s doesn't match master pg_probackup version %s",
remote_agent, PROGRAM_VERSION);
}
fio_communicate(STDIN_FILENO, STDOUT_FILENO);
return 0;
}
else if (strcmp(argv[1], "--help") == 0 ||
strcmp(argv[1], "-?") == 0 ||
strcmp(argv[1], "help") == 0)
@ -352,14 +380,17 @@ main(int argc, char *argv[])
}
canonicalize_path(backup_path);
setMyLocation();
/* Ensure that backup_path is a path to a directory */
rc = fio_stat(backup_path, &stat_buf, true, FIO_BACKUP_HOST);
if (rc != -1 && !S_ISDIR(stat_buf.st_mode))
elog(ERROR, "-B, --backup-path must be a path to directory");
/* Ensure that backup_path is an absolute path */
if (!is_absolute_path(backup_path))
elog(ERROR, "-B, --backup-path must be an absolute path");
/* Ensure that backup_path is a path to a directory */
rc = stat(backup_path, &stat_buf);
if (rc != -1 && !S_ISDIR(stat_buf.st_mode))
elog(ERROR, "-B, --backup-path must be a path to directory");
/* Option --instance is required for all commands except init and show */
if (backup_subcmd != INIT_CMD && backup_subcmd != SHOW_CMD &&
@ -386,7 +417,7 @@ main(int argc, char *argv[])
*/
if (backup_subcmd != INIT_CMD && backup_subcmd != ADD_INSTANCE_CMD)
{
if (access(backup_instance_path, F_OK) != 0)
if (fio_access(backup_instance_path, F_OK, FIO_BACKUP_HOST) != 0)
elog(ERROR, "Instance '%s' does not exist in this backup catalog",
instance_name);
}
@ -411,6 +442,7 @@ main(int argc, char *argv[])
BACKUP_CATALOG_CONF_FILE);
config_read_opt(path, instance_options, ERROR, true, false);
}
setMyLocation();
}
/* Initialize logger */
@ -460,7 +492,10 @@ main(int argc, char *argv[])
elog(ERROR, "Invalid backup-id \"%s\"", backup_id_string);
}
/* Setup stream options. They are used in streamutil.c. */
if (!instance_config.pghost && instance_config.remote.host)
instance_config.pghost = instance_config.remote.host;
/* Setup stream options. They are used in streamutil.c. */
if (instance_config.pghost != NULL)
dbhost = pstrdup(instance_config.pghost);
if (instance_config.pgport != NULL)
@ -520,9 +555,9 @@ main(int argc, char *argv[])
backup_mode = deparse_backup_mode(current.backup_mode);
current.stream = stream_wal;
elog(INFO, "Backup start, pg_probackup version: %s, backup ID: %s, backup mode: %s, instance: %s, stream: %s, remote: %s",
elog(INFO, "Backup start, pg_probackup version: %s, backup ID: %s, backup mode: %s, instance: %s, stream: %s, remote %s",
PROGRAM_VERSION, base36enc(start_time), backup_mode, instance_name,
stream_wal ? "true" : "false", is_remote_backup ? "true" : "false");
stream_wal ? "true" : "false", instance_config.remote.host ? "true" : "false");
return do_backup(start_time, no_validate);
}

62
src/pg_probackup.h
View File

@ -19,16 +19,18 @@
#ifdef FRONTEND
#undef FRONTEND
#include "port/atomics.h"
#include <port/atomics.h>
#define FRONTEND
#else
#include "port/atomics.h"
#include <port/atomics.h>
#endif
#include "utils/configuration.h"
#include "utils/logger.h"
#include "utils/remote.h"
#include "utils/parray.h"
#include "utils/pgut.h"
#include "utils/file.h"
#include "datapagemap.h"
@ -173,6 +175,8 @@ typedef enum ShowFormat
#define BYTES_INVALID (-1) /* file didn`t changed since previous backup, DELTA backup do not rely on it */
#define FILE_NOT_FOUND (-2) /* file disappeared during backup */
#define BLOCKNUM_INVALID (-1)
#define PROGRAM_VERSION "2.0.28"
#define AGENT_PROTOCOL_VERSION 20028
/*
* An instance configuration. It can be stored in a configuration file or passed
@ -202,6 +206,9 @@ typedef struct InstanceConfig
/* Logger parameters */
LoggerConfig logger;
/* Remote access parameters */
RemoteConfig remote;
/* Retention options. 0 disables the option. */
uint32 retention_redundancy;
uint32 retention_window;
@ -312,6 +319,21 @@ typedef struct
int ret;
} backup_files_arg;
/*
* When copying datafiles to backup we validate and compress them block
* by block. Thus special header is required for each data block.
*/
typedef struct BackupPageHeader
{
BlockNumber block; /* block number */
int32 compressed_size;
} BackupPageHeader;
/* Special value for compressed_size field */
#define PageIsTruncated -2
#define SkipCurrentPage -3
/*
* return pointer that exceeds the length of prefix from character string.
* ex. str="/xxx/yyy/zzz", prefix="/xxx/yyy", return="zzz".
@ -351,7 +373,11 @@ typedef struct
XLByteInSeg(xlrp, logSegNo)
#endif
#define IsSshProtocol() (instance_config.remote.host && strcmp(instance_config.remote.proto, "ssh") == 0)
#define IsReplicationProtocol() (instance_config.remote.host && strcmp(instance_config.remote.proto, "replication") == 0)
/* directory options */
extern char *pg_probackup;
extern char *backup_path;
extern char backup_instance_path[MAXPGPATH];
extern char arclog_path[MAXPGPATH];
@ -368,7 +394,9 @@ extern bool temp_slot;
/* backup options */
extern bool smooth_checkpoint;
extern bool is_remote_backup;
/* remote probackup options */
extern char* remote_agent;
extern bool is_ptrack_support;
extern bool is_checksum_enabled;
@ -458,7 +486,8 @@ extern int do_delete_instance(void);
extern char *slurpFile(const char *datadir,
const char *path,
size_t *filesize,
bool safe);
bool safe,
fio_location location);
extern char *fetchFile(PGconn *conn, const char *filename, size_t *filesize);
/* in help.c */
@ -504,6 +533,7 @@ extern bool is_parent(time_t parent_backup_time, pgBackup *child_backup, bool in
extern bool is_prolific(parray *backup_list, pgBackup *target_backup);
extern bool in_backup_list(parray *backup_list, pgBackup *target_backup);
extern int get_backup_index_number(parray *backup_list, pgBackup *backup);
extern bool launch_agent(void);
#define COMPRESS_ALG_DEFAULT NOT_DEFINED_COMPRESS
#define COMPRESS_LEVEL_DEFAULT 1
@ -513,10 +543,12 @@ extern const char* deparse_compress_alg(int alg);
/* in dir.c */
extern void dir_list_file(parray *files, const char *root, bool exclude,
bool omit_symlink, bool add_root, int external_dir_num);
bool omit_symlink, bool add_root, int external_dir_num, fio_location location);
extern void create_data_directories(const char *data_dir,
const char *backup_dir,
bool extract_tablespaces);
bool extract_tablespaces,
fio_location location);
extern void read_tablespace_map(parray *files, const char *backup_dir);
extern void opt_tablespace_map(ConfigOption *opt, const char *arg);
@ -528,7 +560,7 @@ extern char *get_external_remap(char *current_dir);
extern void print_file_list(FILE *out, const parray *files, const char *root,
const char *external_prefix, parray *external_list);
extern parray *dir_read_file_list(const char *root, const char *external_prefix,
const char *file_txt);
const char *file_txt, fio_location location);
extern parray *make_external_directory_list(const char *colon_separated_dirs);
extern void free_dir_list(parray *list);
extern void makeExternalDirPathByNum(char *ret_path, const char *pattern_path,
@ -538,15 +570,15 @@ extern bool backup_contains_external(const char *dir, parray *dirs_list);
extern int dir_create_dir(const char *path, mode_t mode);
extern bool dir_is_empty(const char *path);
extern bool fileExists(const char *path);
extern bool fileExists(const char *path, fio_location location);
extern size_t pgFileSize(const char *path);
extern pgFile *pgFileNew(const char *path, bool omit_symlink, int external_dir_num);
extern pgFile *pgFileNew(const char *path, bool omit_symlink, int external_dir_num, fio_location location);
extern pgFile *pgFileInit(const char *path);
extern void pgFileDelete(pgFile *file);
extern void pgFileFree(void *file);
extern pg_crc32 pgFileGetCRC(const char *file_path, bool use_crc32c,
bool raise_on_deleted, size_t *bytes_read);
bool raise_on_deleted, size_t *bytes_read, fio_location location);
extern int pgFileComparePath(const void *f1, const void *f2);
extern int pgFileComparePathWithExternal(const void *f1, const void *f2);
extern int pgFileComparePathDesc(const void *f1, const void *f2);
@ -564,12 +596,13 @@ extern void restore_data_file(const char *to_path,
pgFile *file, bool allow_truncate,
bool write_header,
uint32 backup_version);
extern bool copy_file(const char *from_root, const char *to_root, pgFile *file);
extern bool copy_file(const char *from_root, fio_location from_location, const char *to_root, fio_location to_location, pgFile *file);
extern void move_file(const char *from_root, const char *to_root, pgFile *file);
extern void push_wal_file(const char *from_path, const char *to_path,
bool is_compress, bool overwrite);
extern void get_wal_file(const char *from_path, const char *to_path);
extern void calc_file_checksum(pgFile *file);
extern void calc_file_checksum(pgFile *file, fio_location location);
extern bool check_file_pages(pgFile *file, XLogRecPtr stop_lsn,
uint32 checksum_version, uint32 backup_version);
@ -602,7 +635,7 @@ extern pg_crc32c get_pgcontrol_checksum(const char *pgdata_path);
extern uint32 get_xlog_seg_size(char *pgdata_path);
extern void set_min_recovery_point(pgFile *file, const char *backup_path,
XLogRecPtr stop_backup_lsn);
extern void copy_pgcontrol_file(const char *from_root, const char *to_root,
extern void copy_pgcontrol_file(const char *from_root, fio_location location, const char *to_root, fio_location to_location,
pgFile *file);
extern void sanityChecks(void);
@ -615,6 +648,9 @@ extern char *base36enc_dup(long unsigned int value);
extern long unsigned int base36dec(const char *text);
extern uint32 parse_server_version(const char *server_version_str);
extern uint32 parse_program_version(const char *program_version);
extern bool parse_page(Page page, XLogRecPtr *lsn);
int32 do_compress(void* dst, size_t dst_size, void const* src, size_t src_size,
CompressAlg alg, int level, const char **errormsg);
#ifdef WIN32
#ifdef _DEBUG

35
src/restore.c
View File

@ -503,7 +503,7 @@ restore_backup(pgBackup *backup, const char *external_dir_str)
* this_backup_path = $BACKUP_PATH/backups/instance_name/backup_id
*/
pgBackupGetPath(backup, this_backup_path, lengthof(this_backup_path), NULL);
create_data_directories(instance_config.pgdata, this_backup_path, true);
create_data_directories(instance_config.pgdata, this_backup_path, true, FIO_DB_HOST);
if(external_dir_str && !skip_external_dirs)
{
@ -526,7 +526,7 @@ restore_backup(pgBackup *backup, const char *external_dir_str)
pgBackupGetPath(backup, external_prefix, lengthof(external_prefix),
EXTERNAL_DIR);
pgBackupGetPath(backup, list_path, lengthof(list_path), DATABASE_FILE_LIST);
files = dir_read_file_list(database_path, external_prefix, list_path);
files = dir_read_file_list(database_path, external_prefix, list_path, FIO_BACKUP_HOST);
/* Restore directories in do_backup_instance way */
parray_qsort(files, pgFileComparePath);
@ -632,12 +632,12 @@ remove_deleted_files(pgBackup *backup)
pgBackupGetPath(backup, filelist_path, lengthof(filelist_path), DATABASE_FILE_LIST);
pgBackupGetPath(backup, external_prefix, lengthof(external_prefix), EXTERNAL_DIR);
/* Read backup's filelist using target database path as base path */
files = dir_read_file_list(instance_config.pgdata, external_prefix, filelist_path);
files = dir_read_file_list(instance_config.pgdata, external_prefix, filelist_path, FIO_BACKUP_HOST);
parray_qsort(files, pgFileComparePathDesc);
/* Get list of files actually existing in target database */
files_restored = parray_new();
dir_list_file(files_restored, instance_config.pgdata, true, true, false, 0);
dir_list_file(files_restored, instance_config.pgdata, true, true, false, 0, FIO_BACKUP_HOST);
/* To delete from leaf, sort in reversed order */
parray_qsort(files_restored, pgFileComparePathDesc);
@ -749,13 +749,17 @@ restore_files(void *arg)
arguments->req_external_dirs))
{
external_path = get_external_remap(external_path);
copy_file(arguments->external_prefix, external_path, file);
copy_file(arguments->external_prefix, FIO_BACKUP_HOST, external_path, FIO_DB_HOST, file);
}
}
else if (strcmp(file->name, "pg_control") == 0)
copy_pgcontrol_file(from_root, instance_config.pgdata, file);
copy_pgcontrol_file(from_root, FIO_BACKUP_HOST,
instance_config.pgdata, FIO_DB_HOST,
file);
else
copy_file(from_root, instance_config.pgdata, file);
copy_file(from_root, FIO_BACKUP_HOST,
instance_config.pgdata, FIO_DB_HOST,
file);
/* print size of restored file */
if (file->write_size != BYTES_INVALID)
@ -793,18 +797,18 @@ create_recovery_conf(time_t backup_id,
elog(LOG, "creating recovery.conf");
snprintf(path, lengthof(path), "%s/recovery.conf", instance_config.pgdata);
fp = fopen(path, "wt");
fp = fio_fopen(path, "wt", FIO_DB_HOST);
if (fp == NULL)
elog(ERROR, "cannot open recovery.conf \"%s\": %s", path,
strerror(errno));
fprintf(fp, "# recovery.conf generated by pg_probackup %s\n",
PROGRAM_VERSION);
fio_fprintf(fp, "# recovery.conf generated by pg_probackup %s\n",
PROGRAM_VERSION);
if (need_restore_conf)
{
fprintf(fp, "restore_command = '%s archive-get -B %s --instance %s "
fio_fprintf(fp, "restore_command = '%s archive-get -B %s --instance %s "
"--wal-file-path %%p --wal-file-name %%f'\n",
PROGRAM_FULL_PATH, backup_path, instance_name);
@ -840,15 +844,14 @@ create_recovery_conf(time_t backup_id,
if (restore_as_replica)
{
fprintf(fp, "standby_mode = 'on'\n");
fio_fprintf(fp, "standby_mode = 'on'\n");
if (backup->primary_conninfo)
fprintf(fp, "primary_conninfo = '%s'\n", backup->primary_conninfo);
fio_fprintf(fp, "primary_conninfo = '%s'\n", backup->primary_conninfo);
}
if (fflush(fp) != 0 ||
fsync(fileno(fp)) != 0 ||
fclose(fp))
if (fio_fflush(fp) != 0 ||
fio_fclose(fp))
elog(ERROR, "cannot write recovery.conf \"%s\": %s", path,
strerror(errno));
}

34
src/util.c
View File

@ -109,7 +109,7 @@ digestControlFile(ControlFileData *ControlFile, char *src, size_t size)
* Write ControlFile to pg_control
*/
static void
writeControlFile(ControlFileData *ControlFile, char *path)
writeControlFile(ControlFileData *ControlFile, char *path, fio_location location)
{
int fd;
char *buffer = NULL;
@ -125,21 +125,19 @@ writeControlFile(ControlFileData *ControlFile, char *path)
memcpy(buffer, ControlFile, sizeof(ControlFileData));
/* Write pg_control */
unlink(path);
fd = open(path,
O_RDWR | O_CREAT | O_EXCL | PG_BINARY,
S_IRUSR | S_IWUSR);
fd = fio_open(path,
O_RDWR | O_CREAT | O_TRUNC | PG_BINARY, location);
if (fd < 0)
elog(ERROR, "Failed to open file: %s", path);
if (write(fd, buffer, ControlFileSize) != ControlFileSize)
if (fio_write(fd, buffer, ControlFileSize) != ControlFileSize)
elog(ERROR, "Failed to overwrite file: %s", path);
if (fsync(fd) != 0)
if (fio_flush(fd) != 0)
elog(ERROR, "Failed to fsync file: %s", path);
close(fd);
fio_close(fd);
pg_free(buffer);
}
@ -156,7 +154,7 @@ get_current_timeline(bool safe)
/* First fetch file... */
buffer = slurpFile(instance_config.pgdata, "global/pg_control", &size,
safe);
safe, FIO_DB_HOST);
if (safe && buffer == NULL)
return 0;
@ -214,7 +212,7 @@ get_system_identifier(const char *pgdata_path)
size_t size;
/* First fetch file... */
buffer = slurpFile(pgdata_path, "global/pg_control", &size, false);
buffer = slurpFile(pgdata_path, "global/pg_control", &size, false, FIO_DB_HOST);
if (buffer == NULL)
return 0;
digestControlFile(&ControlFile, buffer, size);
@ -265,7 +263,7 @@ get_xlog_seg_size(char *pgdata_path)
size_t size;
/* First fetch file... */
buffer = slurpFile(pgdata_path, "global/pg_control", &size, false);
buffer = slurpFile(pgdata_path, "global/pg_control", &size, false, FIO_DB_HOST);
if (buffer == NULL)
return 0;
digestControlFile(&ControlFile, buffer, size);
@ -286,7 +284,7 @@ get_data_checksum_version(bool safe)
/* First fetch file... */
buffer = slurpFile(instance_config.pgdata, "global/pg_control", &size,
safe);
safe, FIO_DB_HOST);
if (buffer == NULL)
return 0;
digestControlFile(&ControlFile, buffer, size);
@ -303,7 +301,7 @@ get_pgcontrol_checksum(const char *pgdata_path)
size_t size;
/* First fetch file... */
buffer = slurpFile(pgdata_path, "global/pg_control", &size, false);
buffer = slurpFile(pgdata_path, "global/pg_control", &size, false, FIO_BACKUP_HOST);
if (buffer == NULL)
return 0;
digestControlFile(&ControlFile, buffer, size);
@ -326,7 +324,7 @@ set_min_recovery_point(pgFile *file, const char *backup_path,
char fullpath[MAXPGPATH];
/* First fetch file content */
buffer = slurpFile(instance_config.pgdata, XLOG_CONTROL_FILE, &size, false);
buffer = slurpFile(instance_config.pgdata, XLOG_CONTROL_FILE, &size, false, FIO_DB_HOST);
if (buffer == NULL)
elog(ERROR, "ERROR");
@ -350,7 +348,7 @@ set_min_recovery_point(pgFile *file, const char *backup_path,
/* overwrite pg_control */
snprintf(fullpath, sizeof(fullpath), "%s/%s", backup_path, XLOG_CONTROL_FILE);
writeControlFile(&ControlFile, fullpath);
writeControlFile(&ControlFile, fullpath, FIO_LOCAL_HOST);
/* Update pg_control checksum in backup_list */
file->crc = ControlFile.crc;
@ -362,14 +360,14 @@ set_min_recovery_point(pgFile *file, const char *backup_path,
* Copy pg_control file to backup. We do not apply compression to this file.
*/
void
copy_pgcontrol_file(const char *from_root, const char *to_root, pgFile *file)
copy_pgcontrol_file(const char *from_root, fio_location from_location, const char *to_root, fio_location to_location, pgFile *file)
{
ControlFileData ControlFile;
char *buffer;
size_t size;
char to_path[MAXPGPATH];
buffer = slurpFile(from_root, XLOG_CONTROL_FILE, &size, false);
buffer = slurpFile(from_root, XLOG_CONTROL_FILE, &size, false, from_location);
digestControlFile(&ControlFile, buffer, size);
@ -378,7 +376,7 @@ copy_pgcontrol_file(const char *from_root, const char *to_root, pgFile *file)
file->write_size = size;
join_path_components(to_path, to_root, file->path + strlen(from_root) + 1);
writeControlFile(&ControlFile, to_path);
writeControlFile(&ControlFile, to_path, to_location);
pg_free(buffer);
}

11
src/utils/configuration.c
View File

@ -8,9 +8,11 @@
*-------------------------------------------------------------------------
*/
#include "pg_probackup.h"
#include "configuration.h"
#include "logger.h"
#include "pgut.h"
#include "file.h"
#include "datatype/timestamp.h"
@ -106,7 +108,7 @@ option_has_arg(char type)
{
case 'b':
case 'B':
return no_argument;
return no_argument;//optional_argument;
default:
return required_argument;
}
@ -504,8 +506,9 @@ config_get_opt(int argc, char **argv, ConfigOption cmd_options[],
if (opt == NULL)
opt = option_find(c, options);
if (opt &&
opt->allowed < SOURCE_CMD && opt->allowed != SOURCE_CMD_STRICT)
if (opt
&& !remote_agent
&& opt->allowed < SOURCE_CMD && opt->allowed != SOURCE_CMD_STRICT)
elog(ERROR, "Option %s cannot be specified in command line",
opt->lname);
/* Check 'opt == NULL' is performed in assign_option() */
@ -567,7 +570,7 @@ config_read_opt(const char *path, ConfigOption options[], int elevel,
}
}
fclose(fp);
fio_close_stream(fp);
return parsed_options;
}

1330
src/utils/file.c Normal file
View File

File diff suppressed because it is too large Load Diff

118
src/utils/file.h Normal file
View File

@ -0,0 +1,118 @@
#ifndef __FILE__H__
#define __FILE__H__
#include "storage/bufpage.h"
#include <stdio.h>
#include <sys/stat.h>
#include <dirent.h>
#ifdef HAVE_LIBZ
#include <zlib.h>
#endif
typedef enum
{
FIO_OPEN,
FIO_CLOSE,
FIO_WRITE,
FIO_RENAME,
FIO_SYMLINK,
FIO_UNLINK,
FIO_MKDIR,
FIO_CHMOD,
FIO_SEEK,
FIO_TRUNCATE,
FIO_PREAD,
FIO_READ,
FIO_LOAD,
FIO_STAT,
FIO_FSTAT,
FIO_SEND,
FIO_ACCESS,
FIO_OPENDIR,
FIO_READDIR,
FIO_CLOSEDIR,
FIO_SEND_PAGES,
FIO_PAGE
} fio_operations;
typedef enum
{
FIO_LOCAL_HOST, /* data is locate at local host */
FIO_DB_HOST, /* data is located at Postgres server host */
FIO_BACKUP_HOST, /* data is located at backup host */
FIO_REMOTE_HOST /* date is located at remote host */
} fio_location;
#define FIO_FDMAX 64
#define FIO_PIPE_MARKER 0x40000000
#define PAGE_CHECKSUM_MISMATCH (-256)
#define SYS_CHECK(cmd) do if ((cmd) < 0) { fprintf(stderr, "%s:%d: (%s) %s\n", __FILE__, __LINE__, #cmd, strerror(errno)); exit(EXIT_FAILURE); } while (0)
#define IO_CHECK(cmd, size) do { int _rc = (cmd); if (_rc != (size)) { if (remote_agent) { fprintf(stderr, "%s:%d: proceeds %d bytes instead of %d: %s\n", __FILE__, __LINE__, _rc, (int)(size), _rc >= 0 ? "end of data" : strerror(errno)); exit(EXIT_FAILURE); } else elog(ERROR, "Communication error: %s", _rc >= 0 ? "end of data" : strerror(errno)); } } while (0)
typedef struct
{
unsigned cop : 5;
unsigned handle : 7;
unsigned size : 20;
unsigned arg;
} fio_header;
extern fio_location MyLocation;
/* Check if FILE handle is local or remote (created by FIO) */
#define fio_is_remote_file(file) ((size_t)(file) <= FIO_FDMAX)
extern void fio_redirect(int in, int out);
extern void fio_communicate(int in, int out);
extern FILE* fio_fopen(char const* name, char const* mode, fio_location location);
extern size_t fio_fwrite(FILE* f, void const* buf, size_t size);
extern ssize_t fio_fread(FILE* f, void* buf, size_t size);
extern int fio_pread(FILE* f, void* buf, off_t offs);
extern int fio_fprintf(FILE* f, char const* arg, ...) __attribute__((format(printf, 2, 3)));
extern int fio_fflush(FILE* f);
extern int fio_fseek(FILE* f, off_t offs);
extern int fio_ftruncate(FILE* f, off_t size);
extern int fio_fclose(FILE* f);
extern int fio_ffstat(FILE* f, struct stat* st);
struct pgFile;
extern int fio_send_pages(FILE* in, FILE* out, struct pgFile *file, XLogRecPtr horizonLsn,
BlockNumber* nBlocksSkipped, int calg, int clevel);
extern int fio_open(char const* name, int mode, fio_location location);
extern ssize_t fio_write(int fd, void const* buf, size_t size);
extern ssize_t fio_read(int fd, void* buf, size_t size);
extern int fio_flush(int fd);
extern int fio_seek(int fd, off_t offs);
extern int fio_fstat(int fd, struct stat* st);
extern int fio_truncate(int fd, off_t size);
extern int fio_close(int fd);
extern int fio_rename(char const* old_path, char const* new_path, fio_location location);
extern int fio_symlink(char const* target, char const* link_path, fio_location location);
extern int fio_unlink(char const* path, fio_location location);
extern int fio_mkdir(char const* path, int mode, fio_location location);
extern int fio_chmod(char const* path, int mode, fio_location location);
extern int fio_access(char const* path, int mode, fio_location location);
extern int fio_stat(char const* path, struct stat* st, bool follow_symlinks, fio_location location);
extern DIR* fio_opendir(char const* path, fio_location location);
extern struct dirent * fio_readdir(DIR *dirp);
extern int fio_closedir(DIR *dirp);
extern FILE* fio_open_stream(char const* name, fio_location location);
extern int fio_close_stream(FILE* f);
#ifdef HAVE_LIBZ
extern gzFile fio_gzopen(char const* path, char const* mode, int level, fio_location location);
extern int fio_gzclose(gzFile file);
extern int fio_gzread(gzFile f, void *buf, unsigned size);
extern int fio_gzwrite(gzFile f, void const* buf, unsigned size);
extern int fio_gzeof(gzFile f);
extern z_off_t fio_gzseek(gzFile f, z_off_t offset, int whence);
extern const char* fio_gzerror(gzFile file, int *errnum);
#endif
#endif

14
src/utils/logger.c
View File

@ -11,6 +11,7 @@
#include <sys/stat.h>
#include "pg_probackup.h"
#include "logger.h"
#include "pgut.h"
#include "thread.h"
@ -127,6 +128,9 @@ exit_if_necessary(int elevel)
pthread_mutex_unlock(&log_file_mutex);
}
if (remote_agent)
sleep(1); /* Let parent receive sent messages */
/* If this is not the main thread then don't call exit() */
if (main_tid != pthread_self())
{
@ -157,12 +161,18 @@ elog_internal(int elevel, bool file_only, const char *message)
time_t log_time = (time_t) time(NULL);
char strfbuf[128];
write_to_file = elevel >= logger_config.log_level_file &&
logger_config.log_directory && logger_config.log_directory[0] != '\0';
write_to_file = elevel >= logger_config.log_level_file
&& logger_config.log_directory
&& logger_config.log_directory[0] != '\0';
write_to_error_log = elevel >= ERROR && logger_config.error_log_filename &&
logger_config.log_directory && logger_config.log_directory[0] != '\0';
write_to_stderr = elevel >= logger_config.log_level_console && !file_only;
if (remote_agent)
{
write_to_stderr |= write_to_error_log | write_to_file;
write_to_error_log = write_to_file = false;
}
pthread_lock(&log_file_mutex);
loggin_in_progress = true;

3
src/utils/pgut.c
View File

@ -19,6 +19,7 @@
#include "pgut.h"
#include "logger.h"
#include "file.h"
const char *PROGRAM_NAME = NULL;
@ -843,7 +844,7 @@ pgut_fopen(const char *path, const char *mode, bool missing_ok)
{
FILE *fp;
if ((fp = fopen(path, mode)) == NULL)
if ((fp = fio_open_stream(path, FIO_BACKUP_HOST)) == NULL)
{
if (missing_ok && errno == ENOENT)
return NULL;

144
src/utils/remote.c Normal file
View File

@ -0,0 +1,144 @@
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <signal.h>
#include "pg_probackup.h"
#include "file.h"
#define MAX_CMDLINE_LENGTH 4096
#define MAX_CMDLINE_OPTIONS 256
#define ERR_BUF_SIZE 4096
static int split_options(int argc, char* argv[], int max_options, char* options)
{
char* opt = options;
char in_quote = '\0';
while (true) {
switch (*opt) {
case '\'':
case '\"':
if (!in_quote) {
in_quote = *opt++;
continue;
}
if (*opt == in_quote && *++opt != in_quote) {
in_quote = '\0';
continue;
}
break;
case '\0':
if (opt != options) {
argv[argc++] = options;
if (argc >= max_options)
elog(ERROR, "Too much options");
}
return argc;
case ' ':
argv[argc++] = options;
if (argc >= max_options)
elog(ERROR, "Too much options");
*opt++ = '\0';
options = opt;
continue;
default:
break;
}
opt += 1;
}
return argc;
}
static int child_pid;
#if 0
static void kill_child(void)
{
kill(child_pid, SIGTERM);
}
#endif
bool launch_agent(void)
{
char cmd[MAX_CMDLINE_LENGTH];
char* ssh_argv[MAX_CMDLINE_OPTIONS];
int ssh_argc;
int outfd[2];
int infd[2];
ssh_argc = 0;
ssh_argv[ssh_argc++] = instance_config.remote.proto;
if (instance_config.remote.port != NULL) {
ssh_argv[ssh_argc++] = "-p";
ssh_argv[ssh_argc++] = instance_config.remote.port;
}
if (instance_config.remote.user != NULL) {
ssh_argv[ssh_argc++] = "-l";
ssh_argv[ssh_argc++] = instance_config.remote.user;
}
if (instance_config.remote.ssh_config != NULL) {
ssh_argv[ssh_argc++] = "-F";
ssh_argv[ssh_argc++] = instance_config.remote.ssh_config;
}
if (instance_config.remote.ssh_options != NULL) {
ssh_argc = split_options(ssh_argc, ssh_argv, MAX_CMDLINE_OPTIONS, pg_strdup(instance_config.remote.ssh_options));
}
if (num_threads > 1)
{
ssh_argv[ssh_argc++] = "-o";
ssh_argv[ssh_argc++] = "PasswordAuthentication=no";
}
ssh_argv[ssh_argc++] = "-o";
ssh_argv[ssh_argc++] = "Compression=no";
ssh_argv[ssh_argc++] = "-o";
ssh_argv[ssh_argc++] = "LogLevel=error";
ssh_argv[ssh_argc++] = instance_config.remote.host;
ssh_argv[ssh_argc++] = cmd;
ssh_argv[ssh_argc] = NULL;
if (instance_config.remote.path)
{
char* sep = strrchr(pg_probackup, '/');
if (sep != NULL) {
pg_probackup = sep + 1;
}
snprintf(cmd, sizeof(cmd), "%s/%s agent %s",
instance_config.remote.path, pg_probackup, PROGRAM_VERSION);
} else {
snprintf(cmd, sizeof(cmd), "%s agent %s", pg_probackup, PROGRAM_VERSION);
}
SYS_CHECK(pipe(infd));
SYS_CHECK(pipe(outfd));
SYS_CHECK(child_pid = fork());
if (child_pid == 0) { /* child */
SYS_CHECK(close(STDIN_FILENO));
SYS_CHECK(close(STDOUT_FILENO));
SYS_CHECK(dup2(outfd[0], STDIN_FILENO));
SYS_CHECK(dup2(infd[1], STDOUT_FILENO));
SYS_CHECK(close(infd[0]));
SYS_CHECK(close(infd[1]));
SYS_CHECK(close(outfd[0]));
SYS_CHECK(close(outfd[1]));
if (execvp(ssh_argv[0], ssh_argv) < 0)
return false;
} else {
elog(LOG, "Spawn agent %d version %s", child_pid, PROGRAM_VERSION);
SYS_CHECK(close(infd[1])); /* These are being used by the child */
SYS_CHECK(close(outfd[0]));
/*atexit(kill_child);*/
fio_redirect(infd[0], outfd[1]); /* write to stdout */
}
return true;
}

24
src/utils/remote.h Normal file
View File

@ -0,0 +1,24 @@
/*-------------------------------------------------------------------------
*
* remote.h: - prototypes of remote functions.
*
* Copyright (c) 2017-2018, Postgres Professional
*
*-------------------------------------------------------------------------
*/
#ifndef REMOTE_H
#define REMOTE_H
typedef struct RemoteConfig
{
char* proto;
char* host;
char* port;
char* path;
char* user;
char *ssh_config;
char *ssh_options;
} RemoteConfig;
#endif

4
src/validate.c
View File

@ -102,7 +102,7 @@ pgBackupValidate(pgBackup *backup)
pgBackupGetPath(backup, base_path, lengthof(base_path), DATABASE_DIR);
pgBackupGetPath(backup, external_prefix, lengthof(external_prefix), EXTERNAL_DIR);
pgBackupGetPath(backup, path, lengthof(path), DATABASE_FILE_LIST);
files = dir_read_file_list(base_path, external_prefix, path);
files = dir_read_file_list(base_path, external_prefix, path, FIO_BACKUP_HOST);
/* setup threads */
for (i = 0; i < parray_num(files); i++)
@ -257,7 +257,7 @@ pgBackupValidateFiles(void *arg)
crc = pgFileGetCRC(file->path,
arguments->backup_version <= 20021 ||
arguments->backup_version >= 20025,
true, NULL);
true, NULL, FIO_LOCAL_HOST);
if (crc != file->crc)
{
elog(WARNING, "Invalid CRC of backup file \"%s\" : %X. Expected %X",

3
tests/Readme.md
View File

@ -19,6 +19,9 @@ Enable compatibility tests:
Specify path to pg_probackup binary file. By default tests use <Path to Git repository>/pg_probackup/
export PGPROBACKUPBIN=<path to pg_probackup>
Remote backup depends on key authentithication to local machine via ssh as current user.
PGPROBACKUP_SSH_REMOTE=ON
Usage:
pip install testgres
pip install psycopg2

24
tests/delta.py
View File

@ -1244,9 +1244,8 @@ class DeltaTest(ProbackupTest, unittest.TestCase):
fname = self.id().split('.')[3]
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
initdb_params=['--data-checksums'],
pg_options={'wal_level': 'replica'}
)
initdb_params=['--data-checksums'])
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
@ -1282,22 +1281,21 @@ class DeltaTest(ProbackupTest, unittest.TestCase):
if self.paranoia:
pgdata = self.pgdata_content(node.data_dir)
log_file_path = os.path.join(backup_dir, "log", "pg_probackup.log")
with open(log_file_path) as f:
self.assertTrue("LOG: File: {0} blknum 1, empty page".format(
file) in f.read())
self.assertFalse("Skipping blknum: 1 in file: {0}".format(
file) in f.read())
if not self.remote:
log_file_path = os.path.join(backup_dir, "log", "pg_probackup.log")
with open(log_file_path) as f:
self.assertTrue("LOG: File: {0} blknum 1, empty page".format(
file) in f.read())
self.assertFalse("Skipping blknum: 1 in file: {0}".format(
file) in f.read())
# Restore DELTA backup
node_restored = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node_restored'),
)
base_dir=os.path.join(module_name, fname, 'node_restored'))
node_restored.cleanup()
self.restore_node(
backup_dir, 'node', node_restored
)
backup_dir, 'node', node_restored)
if self.paranoia:
pgdata_restored = self.pgdata_content(node_restored.data_dir)

2
tests/expected/option_version.out
View File

@ -1 +1 @@
pg_probackup 2.0.27
pg_probackup 2.0.28

65
tests/helpers/ptrack_helpers.py
View File

@ -264,6 +264,34 @@ class ProbackupTest(object):
if self.verbose:
print('PGPROBACKUPBIN_OLD is not an executable file')
self.remote = False
self.remote_host = None
self.remote_port = None
self.remote_user = None
if 'PGPROBACKUP_SSH_REMOTE' in self.test_env:
self.remote = True
# if 'PGPROBACKUP_SSH_HOST' in self.test_env:
# self.remote_host = self.test_env['PGPROBACKUP_SSH_HOST']
# else
# print('PGPROBACKUP_SSH_HOST is not set')
# exit(1)
#
# if 'PGPROBACKUP_SSH_PORT' in self.test_env:
# self.remote_port = self.test_env['PGPROBACKUP_SSH_PORT']
# else
# print('PGPROBACKUP_SSH_PORT is not set')
# exit(1)
#
# if 'PGPROBACKUP_SSH_USER' in self.test_env:
# self.remote_user = self.test_env['PGPROBACKUP_SSH_USER']
# else
# print('PGPROBACKUP_SSH_USER is not set')
# exit(1)
def make_simple_node(
self,
base_dir=None,
@ -607,14 +635,20 @@ class ProbackupTest(object):
except subprocess.CalledProcessError as e:
raise ProbackupException(e.output.decode('utf-8'), command)
def init_pb(self, backup_dir, old_binary=False):
def init_pb(self, backup_dir, options=[], old_binary=False):
shutil.rmtree(backup_dir, ignore_errors=True)
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
return self.run_pb([
'init',
'-B', backup_dir
],
] + options,
old_binary=old_binary
)
@ -627,6 +661,12 @@ class ProbackupTest(object):
'-D', node.data_dir
]
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
return self.run_pb(cmd + options, old_binary=old_binary)
def set_config(self, backup_dir, instance, old_binary=False, options=[]):
@ -675,6 +715,13 @@ class ProbackupTest(object):
'-d', 'postgres',
'--instance={0}'.format(instance)
]
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
if backup_type:
cmd_list += ['-b', backup_type]
@ -696,6 +743,7 @@ class ProbackupTest(object):
self, backup_dir, instance, node=False,
data_dir=None, backup_id=None, old_binary=False, options=[]
):
if data_dir is None:
data_dir = node.data_dir
@ -705,6 +753,13 @@ class ProbackupTest(object):
'-D', data_dir,
'--instance={0}'.format(instance)
]
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
options = options + [
'--remote-proto=ssh',
'--remote-host=localhost']
if backup_id:
cmd_list += ['-i', backup_id]
@ -898,6 +953,10 @@ class ProbackupTest(object):
backup_dir.replace("\\","\\\\"),
instance)
# don`t forget to kill old_binary after remote ssh release
if self.remote and not old_binary:
archive_command = archive_command + '--remote-proto=ssh --remote-host=localhost '
if self.archive_compress or compress:
archive_command = archive_command + '--compress '
@ -1037,6 +1096,8 @@ class ProbackupTest(object):
else:
node.execute('select pg_switch_xlog()')
sleep(1)
def wait_until_replica_catch_with_master(self, master, replica):
if self.version_to_num(

2
tests/merge.py
View File

@ -1242,7 +1242,7 @@ class MergeTest(ProbackupTest, unittest.TestCase):
gdb = self.merge_backup(backup_dir, "node", backup_id, gdb=True)
gdb.set_breakpoint('pgFileDelete')
gdb.set_breakpoint('fio_unlink')
gdb.run_until_break()
gdb.continue_execution_until_break(20)

23
tests/remote.py
View File

@ -17,16 +17,31 @@ class RemoteTest(ProbackupTest, unittest.TestCase):
node = self.make_simple_node(
base_dir=os.path.join(module_name, fname, 'node'),
initdb_params=['--data-checksums'],
pg_options={
'wal_level': 'replica'})
pg_options={'wal_level': 'replica'})
backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
self.init_pb(backup_dir)
self.add_instance(backup_dir, 'node', node)
self.set_archiving(backup_dir, 'node', node)
# self.set_archiving(backup_dir, 'node', node, remote=True)
node.slow_start()
self.backup_node(backup_dir, 'node', node)
self.backup_node(
backup_dir, 'node', node,
options=['--remote-proto=ssh', '--remote-host=localhost', '--stream'])
pgdata = self.pgdata_content(node.data_dir)
node.cleanup()
self.restore_node(
backup_dir, 'node', node,
options=[
'--remote-proto=ssh',
'--remote-host=localhost'])
# Physical comparison
pgdata_restored = self.pgdata_content(node.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
# Clean after yourself
self.del_test_dir(module_name, fname)

20
tests/validate.py
View File

@ -51,14 +51,24 @@ class ValidateTest(ProbackupTest, unittest.TestCase):
self.backup_node(
backup_dir, 'node', node, options=['--log-level-file=verbose'])
log_file_path = os.path.join(backup_dir, "log", "pg_probackup.log")
if self.paranoia:
pgdata = self.pgdata_content(node.data_dir)
with open(log_file_path) as f:
self.assertTrue(
'{0} blknum 1, empty page'.format(file_path) in f.read(),
'Failed to detect nullified block')
if not self.remote:
log_file_path = os.path.join(backup_dir, "log", "pg_probackup.log")
with open(log_file_path) as f:
self.assertTrue(
'{0} blknum 1, empty page'.format(file_path) in f.read(),
'Failed to detect nullified block')
self.validate_pb(backup_dir, options=["-j", "4"])
node.cleanup()
self.restore_node(backup_dir, 'node', node)
if self.paranoia:
pgdata_restored = self.pgdata_content(node.data_dir)
self.compare_pgdata(pgdata, pgdata_restored)
# Clean after yourself
self.del_test_dir(module_name, fname)