database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-30 11:03:19 +03:00
Code Activity

Make pg_basebackup use temporary replication slots

Browse Source 
Temporary replication slots will be used by default when wal streaming
is used and no slot name is specified with -S. If a slot name is
specified, then a permanent slot with that name is used. If --no-slot is
specified, then no permanent or temporary slot will be used.

Temporary slots are only used on 10.0 and newer, of course.
This commit is contained in:
Magnus Hagander
2017-01-16 13:56:43 +01:00
parent 8fa6019b40
commit e7b020f786
9 changed files with 101 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/bin/pg_basebackup/pg_basebackup.c
View File

@ -61,6 +61,11 @@ typedef struct TablespaceList
*/
#define MINIMUM_VERSION_FOR_PG_WAL 100000
/*
* Temporary replication slots are supported from version 10.
*/
#define MINIMUM_VERSION_FOR_TEMP_SLOTS 100000
/*
* Different ways to include WAL
*/
@ -88,6 +93,8 @@ static bool do_sync = true;
static int standby_message_timeout = 10 * 1000; /* 10 sec = default */
static pg_time_t last_progress_report = 0;
static int32 maxrate = 0; /* no limit by default */
static char *replication_slot = NULL;
static bool temp_replication_slot = true;
static bool success = false;
static bool made_new_pgdata = false;
@ -332,6 +339,7 @@ usage(void)
printf(_(" -R, --write-recovery-conf\n"
" write recovery.conf after backup\n"));
printf(_(" -S, --slot=SLOTNAME replication slot to use\n"));
printf(_(" --no-slot prevent creation of temporary replication slot\n"));
printf(_(" -T, --tablespace-mapping=OLDDIR=NEWDIR\n"
" relocate tablespace in OLDDIR to NEWDIR\n"));
printf(_(" -X, --xlog-method=none|fetch|stream\n"
@ -460,6 +468,7 @@ typedef struct
char xlog[MAXPGPATH]; /* directory or tarfile depending on mode */
char *sysidentifier;
int timeline;
bool temp_slot;
} logstreamer_param;
static int
@ -479,6 +488,10 @@ LogStreamerMain(logstreamer_param *param)
stream.do_sync = do_sync;
stream.mark_done = true;
stream.partial_suffix = NULL;
stream.replication_slot = replication_slot;
stream.temp_slot = param->temp_slot;
if (stream.temp_slot && !stream.replication_slot)
stream.replication_slot = psprintf("pg_basebackup_%d", (int) getpid());
if (format == 'p')
stream.walmethod = CreateWalDirectoryMethod(param->xlog, do_sync);
@ -565,6 +578,11 @@ StartLogStreamer(char *startpos, uint32 timeline, char *sysidentifier)
PQserverVersion(conn) < MINIMUM_VERSION_FOR_PG_WAL ?
"pg_xlog" : "pg_wal");
/* Temporary replication slots are only supported in 10 and newer */
if (PQserverVersion(conn) < MINIMUM_VERSION_FOR_TEMP_SLOTS)
param->temp_slot = false;
else
param->temp_slot = temp_replication_slot;
if (format == 'p')
{
@ -2063,11 +2081,13 @@ main(int argc, char **argv)
{"verbose", no_argument, NULL, 'v'},
{"progress", no_argument, NULL, 'P'},
{"xlogdir", required_argument, NULL, 1},
{"no-slot", no_argument, NULL, 2},
{NULL, 0, NULL, 0}
};
int c;
int option_index;
bool no_slot = false;
progname = get_progname(argv[0]);
set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
@ -2117,7 +2137,16 @@ main(int argc, char **argv)
writerecoveryconf = true;
break;
case 'S':
/*
* When specifying replication slot name, use a permanent
* slot.
*/
replication_slot = pg_strdup(optarg);
temp_replication_slot = false;
break;
case 2:
no_slot = true;
break;
case 'T':
tablespace_list_append(optarg);
@ -2277,7 +2306,7 @@ main(int argc, char **argv)
exit(1);
}
if (replication_slot && includewal != STREAM_WAL)
if ((replication_slot || no_slot) && includewal != STREAM_WAL)
{
fprintf(stderr,
_("%s: replication slots can only be used with WAL streaming\n"),
@ -2287,6 +2316,20 @@ main(int argc, char **argv)
exit(1);
}
if (no_slot)
{
if (replication_slot)
{
fprintf(stderr,
_("%s: --no-slot cannot be used with slot name\n"),
progname);
fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
progname);
exit(1);
}
temp_replication_slot = false;
}
if (strcmp(xlog_dir, "") != 0)
{
if (format != 'p')

3
src/bin/pg_basebackup/pg_receivexlog.c
View File

@ -41,6 +41,7 @@ static bool do_create_slot = false;
static bool slot_exists_ok = false;
static bool do_drop_slot = false;
static bool synchronous = false;
static char *replication_slot = NULL;
static void usage(void);
@ -340,6 +341,8 @@ StreamLog(void)
stream.mark_done = false;
stream.walmethod = CreateWalDirectoryMethod(basedir, stream.do_sync);
stream.partial_suffix = ".partial";
stream.replication_slot = replication_slot;
stream.temp_slot = false;
ReceiveXlogStream(conn, &stream);

1
src/bin/pg_basebackup/pg_recvlogical.c
View File

@ -45,6 +45,7 @@ static bool do_create_slot = false;
static bool slot_exists_ok = false;
static bool do_start_slot = false;
static bool do_drop_slot = false;
static char *replication_slot = NULL;
/* filled pairwise with option, value. value may be NULL */
static char **options;

22
src/bin/pg_basebackup/receivelog.c
View File

@ -455,10 +455,10 @@ ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
* synchronous_standby_names, but we've protected them against it so
* far, so let's continue to do so unless specifically requested.
*/
if (replication_slot != NULL)
if (stream->replication_slot != NULL)
{
reportFlushPosition = true;
sprintf(slotcmd, "SLOT \"%s\" ", replication_slot);
sprintf(slotcmd, "SLOT \"%s\" ", stream->replication_slot);
}
else
{
@ -508,6 +508,24 @@ ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
PQclear(res);
}
/*
* Create temporary replication slot if one is needed
*/
if (stream->temp_slot)
{
snprintf(query, sizeof(query),
"CREATE_REPLICATION_SLOT \"%s\" TEMPORARY PHYSICAL RESERVE_WAL",
stream->replication_slot);
res = PQexec(conn, query);
if (PQresultStatus(res) != PGRES_TUPLES_OK)
{
fprintf(stderr, _("%s: could not create temporary replication slot \"%s\": %s"),
progname, stream->replication_slot, PQerrorMessage(conn));
PQclear(res);
return false;
}
}
/*
* initialize flush position to starting point, it's the caller's
* responsibility that that's sane.

6
src/bin/pg_basebackup/receivelog.h
View File

@ -37,13 +37,15 @@ typedef struct StreamCtl
* often */
bool synchronous; /* Flush immediately WAL data on write */
bool mark_done; /* Mark segment as done in generated archive */
bool do_sync; /* Flush to disk to ensure consistent state
* of data */
bool do_sync; /* Flush to disk to ensure consistent state of
* data */
stream_stop_callback stream_stop; /* Stop streaming when returns true */
WalWriteMethod *walmethod; /* How to write the WAL */
char *partial_suffix; /* Suffix appended to partially received files */
char *replication_slot; /* Replication slot to use, or NULL */
bool temp_slot; /* Create temporary replication slot */
} StreamCtl;

1
src/bin/pg_basebackup/streamutil.c
View File

@ -38,7 +38,6 @@ char *connection_string = NULL;
char *dbhost = NULL;
char *dbuser = NULL;
char *dbport = NULL;
char *replication_slot = NULL;
char *dbname = NULL;
int dbgetpassword = 0; /* 0=auto, -1=never, 1=always */
static bool have_password = false;

1
src/bin/pg_basebackup/streamutil.h
View File

@ -23,7 +23,6 @@ extern char *dbuser;
extern char *dbport;
extern char *dbname;
extern int dbgetpassword;
extern char *replication_slot;
/* Connection kept global so we can disconnect easily */
extern PGconn *conn;

5
src/bin/pg_basebackup/t/010_pg_basebackup.pl
View File

@ -4,7 +4,7 @@ use Cwd;
use Config;
use PostgresNode;
use TestLib;
use Test::More tests => 71;
use Test::More tests => 72;
program_help_ok('pg_basebackup');
program_version_ok('pg_basebackup');
@ -244,6 +244,9 @@ $node->command_ok(
[ 'pg_basebackup', '-D', "$tempdir/backupxst", '-X', 'stream', '-Ft' ],
'pg_basebackup -X stream runs in tar mode');
ok(-f "$tempdir/backupxst/pg_wal.tar", "tar file was created");
$node->command_ok(
[ 'pg_basebackup', '-D', "$tempdir/backupnoslot", '-X', 'stream', '--no-slot' ],
'pg_basebackup -X stream runs with --no-slot');
$node->command_fails(
[ 'pg_basebackup', '-D', "$tempdir/fail", '-S', 'slot1' ],