Pgpro 5421. --wal-file-path option

doc/pgprobackup.xml

@@ -131,6 +131,7 @@ doc/src/sgml/pgprobackup.sgml
    <arg choice="plain"><option>archive-push</option></arg>
    <arg choice="plain"><option>-B</option> <replaceable>backup_dir</replaceable></arg>
    <arg choice="plain"><option>--instance</option> <replaceable>instance_name</replaceable></arg>
+   <arg choice="plain"><option>--wal-file-path</option> <replaceable>wal_file_path</replaceable></arg>
    <arg choice="plain"><option>--wal-file-name</option> <replaceable>wal_file_name</replaceable></arg>
    <arg rep="repeat"><replaceable>option</replaceable></arg>
   </cmdsynopsis>
@@ -5367,7 +5368,9 @@ pg_probackup catchup -b <replaceable>catchup_mode</replaceable>
         Provides the path to the WAL file in
         <parameter>archive_command</parameter> and
         <parameter>restore_command</parameter>. Use the <literal>%p</literal>
-        variable as the value for this option for correct processing.
+        variable as the value for this option or explicitly specify the path to a file
+        outside of the data directory. If you skip this option, the path
+        specified in <filename>pg_probackup.conf</filename> will be used.
       </para>
       </listitem>
       </varlistentry>
@@ -5380,6 +5383,8 @@ pg_probackup catchup -b <replaceable>catchup_mode</replaceable>
         <parameter>archive_command</parameter> and
         <parameter>restore_command</parameter>. Use the <literal>%f</literal>
         variable as the value for this option for correct processing.
+        If the value of <option>--wal-file-path</option> is a path
+        outside of the data directory, explicitly specify the filename.
       </para>
       </listitem>
       </varlistentry>

src/archive.c

@@ -3,7 +3,7 @@
  * archive.c: -  pg_probackup specific archive commands for archive backups.
  *
  *
- * Portions Copyright (c) 2018-2019, Postgres Professional
+ * Portions Copyright (c) 2018-2021, Postgres Professional
  *
  *-------------------------------------------------------------------------
  */
@@ -113,15 +113,13 @@ static parray *setup_push_filelist(const char *archive_status_dir,
  * Where archlog_path is $BACKUP_PATH/wal/instance_name
  */
 void
-do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wal_file_path,
+do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *pg_xlog_dir,
 				char *wal_file_name, int batch_size, bool overwrite,
 				bool no_sync, bool no_ready_rename)
 {
 	uint64		i;
-	char		current_dir[MAXPGPATH];
-	char		pg_xlog_dir[MAXPGPATH];
-	char		archive_status_dir[MAXPGPATH];
-	uint64		system_id;
+	/* usually instance pgdata/pg_wal/archive_status, empty if no_ready_rename or batch_size == 1 */
+	char		archive_status_dir[MAXPGPATH] = "";
 	bool		is_compress = false;
 
 	/* arrays with meta info for multi threaded backup */
@@ -141,31 +139,8 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
 	parray     *batch_files = NULL;
 	int         n_threads;
 
-	if (wal_file_name == NULL)
-		elog(ERROR, "Required parameter is not specified: --wal-file-name %%f");
-
-	if (!getcwd(current_dir, sizeof(current_dir)))
-		elog(ERROR, "getcwd() error");
-
-	/* verify that archive-push --instance parameter is valid */
-	system_id = get_system_identifier(current_dir, FIO_DB_HOST);
-
-	if (instance->pgdata == NULL)
-		elog(ERROR, "Cannot read pg_probackup.conf for this instance");
-
-	if (system_id != instance->system_identifier)
-		elog(ERROR, "Refuse to push WAL segment %s into archive. Instance parameters mismatch."
-					"Instance '%s' should have SYSTEM_ID = " UINT64_FORMAT " instead of " UINT64_FORMAT,
-				wal_file_name, instanceState->instance_name, instance->system_identifier, system_id);
-
-	if (instance->compress_alg == PGLZ_COMPRESS)
-		elog(ERROR, "Cannot use pglz for WAL compression");
-
-	join_path_components(pg_xlog_dir, current_dir, XLOGDIR);
-	join_path_components(archive_status_dir, pg_xlog_dir, "archive_status");
-
-	/* Create 'archlog_path' directory. Do nothing if it already exists. */
-	//fio_mkdir(instanceState->instance_wal_subdir_path, DIR_PERMISSION, FIO_BACKUP_HOST);
+	if (!no_ready_rename || batch_size > 1)
+		join_path_components(archive_status_dir, pg_xlog_dir, "archive_status");
 
 #ifdef HAVE_LIBZ
 	if (instance->compress_alg == ZLIB_COMPRESS)
@@ -204,12 +179,13 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
 		{
 			int rc;
 			WALSegno *xlogfile = (WALSegno *) parray_get(batch_files, i);
+			bool first_wal = strcmp(xlogfile->name, wal_file_name) == 0;
 
-			rc = push_file(xlogfile, archive_status_dir,
+			rc = push_file(xlogfile, first_wal ? NULL : archive_status_dir,
 						   pg_xlog_dir, instanceState->instance_wal_subdir_path,
 						   overwrite, no_sync,
 						   instance->archive_timeout,
-						   no_ready_rename || (strcmp(xlogfile->name, wal_file_name) == 0) ? true : false,
+						   no_ready_rename || first_wal,
 						   is_compress && IsXLogFileName(xlogfile->name) ? true : false,
 						   instance->compress_level);
 			if (rc == 0)
@@ -233,7 +209,7 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
 		arg->first_filename = wal_file_name;
 		arg->archive_dir = instanceState->instance_wal_subdir_path;
 		arg->pg_xlog_dir = pg_xlog_dir;
-		arg->archive_status_dir = archive_status_dir;
+		arg->archive_status_dir = (!no_ready_rename || batch_size > 1) ? archive_status_dir : NULL;
 		arg->overwrite = overwrite;
 		arg->compress = is_compress;
 		arg->no_sync = no_sync;
@@ -276,7 +252,7 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
 
 	/* Note, that we are leaking memory here,
 	 * because pushing into archive is a very
-	 * time-sensetive operation, so we skip freeing stuff.
+	 * time-sensitive operation, so we skip freeing stuff.
 	 */
 
 push_done:
@@ -356,9 +332,6 @@ push_file(WALSegno *xlogfile, const char *archive_status_dir,
 		  int compress_level)
 {
 	int     rc;
-	char	wal_file_dummy[MAXPGPATH];
-
-	join_path_components(wal_file_dummy, archive_status_dir, xlogfile->name);
 
 	elog(LOG, "pushing file \"%s\"", xlogfile->name);
 
@@ -375,11 +348,13 @@ push_file(WALSegno *xlogfile, const char *archive_status_dir,
 #endif
 
 	/* take '--no-ready-rename' flag into account */
-	if (!no_ready_rename)
+	if (!no_ready_rename && archive_status_dir != NULL)
 	{
+		char	wal_file_dummy[MAXPGPATH];
 		char	wal_file_ready[MAXPGPATH];
 		char	wal_file_done[MAXPGPATH];
 
+		join_path_components(wal_file_dummy, archive_status_dir, xlogfile->name);
 		snprintf(wal_file_ready, MAXPGPATH, "%s.%s", wal_file_dummy, "ready");
 		snprintf(wal_file_done, MAXPGPATH, "%s.%s", wal_file_dummy, "done");
 

src/backup.c

@@ -943,7 +943,7 @@ check_system_identifiers(PGconn *conn, const char *pgdata)
 	uint64		system_id_conn;
 	uint64		system_id_pgdata;
 
-	system_id_pgdata = get_system_identifier(pgdata, FIO_DB_HOST);
+	system_id_pgdata = get_system_identifier(pgdata, FIO_DB_HOST, false);
 	system_id_conn = get_remote_system_identifier(conn);
 
 	/* for checkdb check only system_id_pgdata and system_id_conn */

src/catchup.c

@@ -48,7 +48,7 @@ catchup_init_state(PGNodeInfo	*source_node_info, const char *source_pgdata, cons
 
 	/* Get WAL segments size and system ID of source PG instance */
 	instance_config.xlog_seg_size = get_xlog_seg_size(source_pgdata);
-	instance_config.system_identifier = get_system_identifier(source_pgdata, FIO_DB_HOST);
+	instance_config.system_identifier = get_system_identifier(source_pgdata, FIO_DB_HOST, false);
 	current.start_time = time(NULL);
 
 	strlcpy(current.program_version, PROGRAM_VERSION, sizeof(current.program_version));
@@ -163,15 +163,15 @@ catchup_preflight_checks(PGNodeInfo *source_node_info, PGconn *source_conn,
 		uint64	source_conn_id, source_id, dest_id;
 
 		source_conn_id = get_remote_system_identifier(source_conn);
-		source_id = get_system_identifier(source_pgdata, FIO_DB_HOST); /* same as instance_config.system_identifier */
+		source_id = get_system_identifier(source_pgdata, FIO_DB_HOST, false); /* same as instance_config.system_identifier */
 
 		if (source_conn_id != source_id)
 			elog(ERROR, "Database identifiers mismatch: we connected to DB id %lu, but in \"%s\" we found id %lu",
 				source_conn_id, source_pgdata, source_id);
 
 		if (current.backup_mode != BACKUP_MODE_FULL)
 		{
-			dest_id = get_system_identifier(dest_pgdata, FIO_LOCAL_HOST);
+			dest_id = get_system_identifier(dest_pgdata, FIO_LOCAL_HOST, false);
 			if (source_conn_id != dest_id)
 			elog(ERROR, "Database identifiers mismatch: we connected to DB id %lu, but in \"%s\" we found id %lu",
 				source_conn_id, dest_pgdata, dest_id);

src/help.c

@@ -227,6 +227,7 @@ help_pg_probackup(void)
 
 	printf(_("\n  %s archive-push -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
 	printf(_("                 --wal-file-name=wal-file-name\n"));
+	printf(_("                 [--wal-file-path=wal-file-path]\n"));
 	printf(_("                 [-j num-threads] [--batch-size=batch_size]\n"));
 	printf(_("                 [--archive-timeout=timeout]\n"));
 	printf(_("                 [--no-ready-rename] [--no-sync]\n"));
@@ -937,6 +938,7 @@ help_archive_push(void)
 {
 	printf(_("\n%s archive-push -B backup-path --instance=instance_name\n"), PROGRAM_NAME);
 	printf(_("                 --wal-file-name=wal-file-name\n"));
+	printf(_("                 [--wal-file-path=wal-file-path]\n"));
 	printf(_("                 [-j num-threads] [--batch-size=batch_size]\n"));
 	printf(_("                 [--archive-timeout=timeout]\n"));
 	printf(_("                 [--no-ready-rename] [--no-sync]\n"));
@@ -951,6 +953,8 @@ help_archive_push(void)
 	printf(_("      --instance=instance_name     name of the instance to delete\n"));
 	printf(_("      --wal-file-name=wal-file-name\n"));
 	printf(_("                                   name of the file to copy into WAL archive\n"));
+	printf(_("      --wal-file-path=wal-file-path\n"));
+	printf(_("                                   relative destination path of the WAL archive\n"));
 	printf(_("  -j, --threads=NUM                number of parallel threads\n"));
 	printf(_("      --batch-size=NUM             number of files to be copied\n"));
 	printf(_("      --archive-timeout=timeout    wait timeout before discarding stale temp file(default: 5min)\n"));
@@ -981,8 +985,8 @@ static void
 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"));
+	printf(_("                 [--wal-file-path=wal-file-path]\n"));
 	printf(_("                 [-j num-threads] [--batch-size=batch_size]\n"));
 	printf(_("                 [--no-validate-wal]\n"));
 	printf(_("                 [--remote-proto] [--remote-host]\n"));

src/init.c

@@ -57,7 +57,7 @@ do_add_instance(InstanceState *instanceState, InstanceConfig *instance)
 						 "(-D, --pgdata)");
 
 	/* Read system_identifier from PGDATA */
-	instance->system_identifier = get_system_identifier(instance->pgdata, FIO_DB_HOST);
+	instance->system_identifier = get_system_identifier(instance->pgdata, FIO_DB_HOST, false);
 	/* Starting from PostgreSQL 11 read WAL segment size from PGDATA */
 	instance->xlog_seg_size = get_xlog_seg_size(instance->pgdata);
 

src/pg_probackup.c

@@ -35,7 +35,7 @@
  * which includes info about pgdata directory and connection.
  *
  * Portions Copyright (c) 2009-2013, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
- * Portions Copyright (c) 2015-2019, Postgres Professional
+ * Portions Copyright (c) 2015-2021, Postgres Professional
  *
  *-------------------------------------------------------------------------
  */
@@ -151,6 +151,7 @@ static char *wal_file_path;
 static char *wal_file_name;
 static bool file_overwrite = false;
 static bool no_ready_rename = false;
+static char archive_push_xlog_dir[MAXPGPATH] = "";
 
 /* archive get options */
 static char *prefetch_dir;
@@ -788,14 +789,105 @@ main(int argc, char *argv[])
 		current.stream = stream_wal = true;
 		if (instance_config.external_dir_str)
 			elog(ERROR, "external directories not supported fom \"%s\" command", get_subcmd_name(backup_subcmd));
-		// TODO проверить instance_config.conn_opt
+		// TODO check instance_config.conn_opt
 	}
 
 	/* sanity */
 	if (backup_subcmd == VALIDATE_CMD && restore_params->no_validate)
 		elog(ERROR, "You cannot specify \"--no-validate\" option with the \"%s\" command",
 			get_subcmd_name(backup_subcmd));
 
+	if (backup_subcmd == ARCHIVE_PUSH_CMD)
+	{
+		/* Check archive-push parameters and construct archive_push_xlog_dir
+		 *
+		 * There are 4 cases:
+		 * 1. no --wal-file-path specified -- use cwd, ./PG_XLOG_DIR for wal files
+		 * (and ./PG_XLOG_DIR/archive_status for .done files inside do_archive_push())
+		 * in this case we can use batches and threads
+		 * 2. --wal-file-path is specified and it is the same dir as stored in pg_probackup.conf (instance_config.pgdata)
+		 * in this case we can use this path, as well as batches and thread
+		 * 3. --wal-file-path is specified and it isn't same dir as stored in pg_probackup.conf but control file present with correct system_id
+		 * in this case we can use this path, as well as batches and thread
+		 * (replica for example, see test_archive_push_sanity)
+		 * 4. --wal-file-path is specified and it is different from instance_config.pgdata and no control file found
+		 * disable optimizations and work with user specified path
+		 */
+		bool	check_system_id = true;
+		uint64	system_id;
+		char	current_dir[MAXPGPATH];
+
+		if (wal_file_name == NULL)
+			elog(ERROR, "Required parameter is not specified: --wal-file-name %%f");
+
+		if (instance_config.pgdata == NULL)
+			elog(ERROR, "Cannot read pg_probackup.conf for this instance");
+
+		/* TODO may be remove in preference of checking inside compress_init()? */
+		if (instance_config.compress_alg == PGLZ_COMPRESS)
+                        elog(ERROR, "Cannot use pglz for WAL compression");
+
+		if (!getcwd(current_dir, sizeof(current_dir)))
+			elog(ERROR, "getcwd() error");
+
+		if (wal_file_path == NULL)
+		{
+			/* 1st case */
+			system_id = get_system_identifier(current_dir, FIO_DB_HOST, false);
+			join_path_components(archive_push_xlog_dir, current_dir, XLOGDIR);
+		}
+		else
+		{
+			/*
+			 * Usually we get something like
+			 *   wal_file_path = "pg_wal/0000000100000000000000A1"
+			 *   wal_file_name = "0000000100000000000000A1"
+			 *   instance_config.pgdata = "/pgdata/.../node/data"
+			 * We need to strip wal_file_name from wal_file_path, add XLOGDIR to instance_config.pgdata
+			 * and compare this directories.
+			 * Note, that pg_wal can be symlink (see test_waldir_outside_pgdata_archiving)
+			 */
+			char	*stripped_wal_file_path = pgut_str_strip_trailing_filename(wal_file_path, wal_file_name);
+			join_path_components(archive_push_xlog_dir, instance_config.pgdata, XLOGDIR);
+			if (fio_is_same_file(stripped_wal_file_path, archive_push_xlog_dir, true, FIO_DB_HOST))
+			{
+				/* 2nd case */
+				system_id = get_system_identifier(instance_config.pgdata, FIO_DB_HOST, false);
+				/* archive_push_xlog_dir already have right value */
+			}
+			else
+			{
+				if (strlen(stripped_wal_file_path) < MAXPGPATH)
+					strncpy(archive_push_xlog_dir, stripped_wal_file_path, MAXPGPATH);
+				else
+					elog(ERROR, "Value specified to --wal_file_path is too long");
+
+				system_id = get_system_identifier(current_dir, FIO_DB_HOST, true);
+				/* 3rd case if control file present -- i.e. system_id != 0 */
+
+				if (system_id == 0)
+				{
+					/* 4th case */
+					check_system_id = false;
+
+					if (batch_size > 1 || num_threads > 1 || !no_ready_rename)
+					{
+						elog(WARNING, "Supplied --wal_file_path is outside pgdata, force safe values for options: --batch-size=1 -j 1 --no-ready-rename");
+						batch_size = 1;
+						num_threads = 1;
+						no_ready_rename = true;
+					}
+				}
+			}
+			pfree(stripped_wal_file_path);
+		}
+
+		if (check_system_id && system_id != instance_config.system_identifier)
+			elog(ERROR, "Refuse to push WAL segment %s into archive. Instance parameters mismatch."
+						"Instance '%s' should have SYSTEM_ID = " UINT64_FORMAT " instead of " UINT64_FORMAT,
+					wal_file_name, instanceState->instance_name, instance_config.system_identifier, system_id);
+	}
+
 #if PG_VERSION_NUM >= 100000
 	if (temp_slot && perm_slot)
 		elog(ERROR, "You cannot specify \"--perm-slot\" option with the \"--temp-slot\" option");
@@ -819,7 +911,7 @@ main(int argc, char *argv[])
 	switch (backup_subcmd)
 	{
 		case ARCHIVE_PUSH_CMD:
-			do_archive_push(instanceState, &instance_config, wal_file_path, wal_file_name,
+			do_archive_push(instanceState, &instance_config, archive_push_xlog_dir, wal_file_name,
 							batch_size, file_overwrite, no_sync, no_ready_rename);
 			break;
 		case ARCHIVE_GET_CMD:

src/pg_probackup.h

@@ -889,7 +889,7 @@ extern int do_init(CatalogState *catalogState);
 extern int do_add_instance(InstanceState *instanceState, InstanceConfig *instance);
 
 /* in archive.c */
-extern void do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wal_file_path,
+extern void do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *pg_xlog_dir,
 						   char *wal_file_name, int batch_size, bool overwrite,
 						   bool no_sync, bool no_ready_rename);
 extern void do_archive_get(InstanceState *instanceState, InstanceConfig *instance, const char *prefetch_dir_arg, char *wal_file_path,
@@ -1153,7 +1153,7 @@ extern XLogRecPtr get_next_record_lsn(const char *archivedir, XLogSegNo	segno, T
 extern TimeLineID get_current_timeline(PGconn *conn);
 extern TimeLineID get_current_timeline_from_control(const char *pgdata_path, fio_location location, bool safe);
 extern XLogRecPtr get_checkpoint_location(PGconn *conn);
-extern uint64 get_system_identifier(const char *pgdata_path, fio_location location);
+extern uint64 get_system_identifier(const char *pgdata_path, fio_location location, bool safe);
 extern uint64 get_remote_system_identifier(PGconn *conn);
 extern uint32 get_data_checksum_version(bool safe);
 extern pg_crc32c get_pgcontrol_checksum(const char *pgdata_path);

src/restore.c

@@ -2186,7 +2186,7 @@ check_incremental_compatibility(const char *pgdata, uint64 system_identifier,
 	 */
 	elog(INFO, "Trying to read pg_control file in destination directory");
 
-	system_id_pgdata = get_system_identifier(pgdata, FIO_DB_HOST);
+	system_id_pgdata = get_system_identifier(pgdata, FIO_DB_HOST, false);
 
 	if (system_id_pgdata == instance_config.system_identifier)
 		system_id_match = true;