[Issue #449] subdirectories for WAL archive

Author: gsmolk

src/archive.c

@@ -15,14 +15,17 @@
 
 static int push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_dir,
 								  const char *archive_dir, bool overwrite, bool no_sync,
-								  uint32 archive_timeout);
+								  uint32 archive_timeout, xlogFileType type);
 #ifdef HAVE_LIBZ
 static int push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
-									 const char *archive_dir, bool overwrite, bool no_sync,
-									 int compress_level, uint32 archive_timeout);
+								 const char *archive_dir, bool overwrite, bool no_sync,
+								 int compress_level, uint32 archive_timeout, xlogFileType type);
 #endif
 static void *push_files(void *arg);
 static void *get_files(void *arg);
+static bool
+get_wal_file_wrapper(const char *filename, const char *archive_root_dir,
+					 const char *to_fullpath, bool prefetch_mode);
 static bool get_wal_file(const char *filename, const char *from_path, const char *to_path,
 													bool prefetch_mode);
 static int get_wal_file_internal(const char *from_path, const char *to_path, FILE *out,
@@ -89,8 +92,9 @@ typedef struct
 
 typedef struct WALSegno
 {
-	char        name[MAXFNAMELEN];
-	volatile    pg_atomic_flag lock;
+	char         name[MAXFNAMELEN];
+	volatile     pg_atomic_flag lock;
+	xlogFileType type;
 } WALSegno;
 
 static int push_file(WALSegno *xlogfile, const char *archive_status_dir,
@@ -102,6 +106,28 @@ static int push_file(WALSegno *xlogfile, const char *archive_status_dir,
 static parray *setup_push_filelist(const char *archive_status_dir,
 								   const char *first_file, int batch_size);
 
+static xlogFileType
+get_xlogFileType(const char *filename)
+{
+
+	if IsXLogFileName(filename)
+		return SEGMENT;
+
+	else if IsPartialXLogFileName(filename)
+		return PARTIAL_SEGMENT;
+
+	else if IsBackupHistoryFileName(filename)
+		return BACKUP_HISTORY_FILE;
+
+	else if IsTLHistoryFileName(filename)
+		return HISTORY_FILE;
+
+	else if IsBackupHistoryFileName(filename)
+		return BACKUP_HISTORY_FILE;
+
+	return UNKNOWN;
+}
+
 /*
  * At this point, we already done one roundtrip to archive server
  * to get instance config.
@@ -185,6 +211,8 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
 						parray_num(batch_files), batch_size,
 						is_compress ? "zlib" : "none");
 
+	/* TODO: create subdirectories here, not in internal functions */
+
 	num_threads = n_threads;
 
 	/* Single-thread push
@@ -366,12 +394,12 @@ push_file(WALSegno *xlogfile, const char *archive_status_dir,
 	if (!is_compress)
 		rc = push_file_internal_uncompressed(xlogfile->name, pg_xlog_dir,
 											 archive_dir, overwrite, no_sync,
-											 archive_timeout);
+											 archive_timeout, xlogfile->type);
 #ifdef HAVE_LIBZ
 	else
 		rc = push_file_internal_gz(xlogfile->name, pg_xlog_dir, archive_dir,
 								   overwrite, no_sync, compress_level,
-								   archive_timeout);
+								   archive_timeout, xlogfile->type);
 #endif
 
 	/* take '--no-ready-rename' flag into account */
@@ -408,13 +436,14 @@ push_file(WALSegno *xlogfile, const char *archive_status_dir,
 int
 push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_dir,
 								const char *archive_dir, bool overwrite, bool no_sync,
-								uint32 archive_timeout)
+								uint32 archive_timeout, xlogFileType type)
 {
 	FILE	   *in = NULL;
 	int			out = -1;
 	char       *buf = pgut_malloc(OUT_BUF_SIZE); /* 1MB buffer */
 	char		from_fullpath[MAXPGPATH];
 	char		to_fullpath[MAXPGPATH];
+	char        archive_subdir[MAXPGPATH];
 	/* partial handling */
 	struct stat		st;
 	char		to_fullpath_part[MAXPGPATH];
@@ -427,8 +456,16 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
 	/* from path */
 	join_path_components(from_fullpath, pg_xlog_dir, wal_file_name);
 	canonicalize_path(from_fullpath);
+
+	/* calculate subdir in WAL archive */
+	get_archive_subdir(archive_subdir, archive_dir, wal_file_name, type);
+
+	/* create subdirectory */
+	if (fio_mkdir(archive_subdir, DIR_PERMISSION, FIO_BACKUP_HOST) != 0)
+		elog(ERROR, "Cannot create subdirectory in WAL archive: '%s'", archive_subdir);
+
 	/* to path */
-	join_path_components(to_fullpath, archive_dir, wal_file_name);
+	join_path_components(to_fullpath, archive_subdir, wal_file_name);
 	canonicalize_path(to_fullpath);
 
 	/* Open source file for read */
@@ -647,14 +684,15 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
 int
 push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
 					  const char *archive_dir, bool overwrite, bool no_sync,
-					  int compress_level, uint32 archive_timeout)
+					  int compress_level, uint32 archive_timeout, xlogFileType type)
 {
 	FILE	   *in = NULL;
 	gzFile		out = NULL;
 	char       *buf = pgut_malloc(OUT_BUF_SIZE);
 	char		from_fullpath[MAXPGPATH];
 	char		to_fullpath[MAXPGPATH];
 	char		to_fullpath_gz[MAXPGPATH];
+	char        archive_subdir[MAXPGPATH];
 
 	/* partial handling */
 	struct stat		st;
@@ -669,8 +707,16 @@ push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
 	/* from path */
 	join_path_components(from_fullpath, pg_xlog_dir, wal_file_name);
 	canonicalize_path(from_fullpath);
+
+	/* calculate subdir in WAL archive */
+	get_archive_subdir(archive_subdir, archive_dir, wal_file_name, type);
+
+	/* create subdirectory */
+	if (fio_mkdir(archive_subdir, DIR_PERMISSION, FIO_BACKUP_HOST) != 0)
+		elog(ERROR, "Cannot create subdirectory in WAL archive: '%s'", archive_subdir);
+
 	/* to path */
-	join_path_components(to_fullpath, archive_dir, wal_file_name);
+	join_path_components(to_fullpath, archive_subdir, wal_file_name);
 	canonicalize_path(to_fullpath);
 
 	/* destination file with .gz suffix */
@@ -940,15 +986,17 @@ setup_push_filelist(const char *archive_status_dir, const char *first_file,
 {
 	int i;
 	WALSegno *xlogfile = NULL;
-	parray  *status_files = NULL;
-	parray  *batch_files = parray_new();
+	parray   *status_files = NULL;
+	parray   *batch_files = parray_new();
 
 	/* guarantee that first filename is in batch list */
 	xlogfile = palloc(sizeof(WALSegno));
 	pg_atomic_init_flag(&xlogfile->lock);
 	snprintf(xlogfile->name, MAXFNAMELEN, "%s", first_file);
 	parray_append(batch_files, xlogfile);
 
+	xlogfile->type = get_xlogFileType(xlogfile->name);
+
 	if (batch_size < 2)
 		return batch_files;
 
@@ -980,6 +1028,8 @@ setup_push_filelist(const char *archive_status_dir, const char *first_file,
 		pg_atomic_init_flag(&xlogfile->lock);
 
 		snprintf(xlogfile->name, MAXFNAMELEN, "%s", filename);
+
+		xlogfile->type = get_xlogFileType(xlogfile->name);
 		parray_append(batch_files, xlogfile);
 
 		if (parray_num(batch_files) >= batch_size)
@@ -1048,7 +1098,7 @@ do_archive_get(InstanceState *instanceState, InstanceConfig *instance, const cha
 
 	/* full filepath to WAL file in archive directory.
 	 * $BACKUP_PATH/wal/instance_name/000000010000000000000001 */
-	join_path_components(backup_wal_file_path, instanceState->instance_wal_subdir_path, wal_file_name);
+	//join_path_components(backup_wal_file_path, instanceState->instance_wal_subdir_path, wal_file_name);
 
 	INSTR_TIME_SET_CURRENT(start_time);
 	if (num_threads > batch_size)
@@ -1177,7 +1227,7 @@ do_archive_get(InstanceState *instanceState, InstanceConfig *instance, const cha
 
 	while (fail_count < 3)
 	{
-		if (get_wal_file(wal_file_name, backup_wal_file_path, absolute_wal_file_path, false))
+		if (get_wal_file_wrapper(wal_file_name, instanceState->instance_wal_subdir_path, absolute_wal_file_path, false))
 		{
 			fail_count = 0;
 			elog(INFO, "pg_probackup archive-get copied WAL file %s", wal_file_name);
@@ -1260,7 +1310,7 @@ uint32 run_wal_prefetch(const char *prefetch_dir, const char *archive_dir,
 			/* It is ok, maybe requested batch is greater than the number of available
 			 * files in the archive
 			 */
-			if (!get_wal_file(xlogfile->name, from_fullpath, to_fullpath, true))
+			if (!get_wal_file_wrapper(xlogfile->name, archive_dir, to_fullpath, true))
 			{
 				elog(LOG, "Thread [%d]: Failed to prefetch WAL segment %s", 0, xlogfile->name);
 				break;
@@ -1334,7 +1384,7 @@ get_files(void *arg)
 		join_path_components(from_fullpath, args->archive_dir, xlogfile->name);
 		join_path_components(to_fullpath, args->prefetch_dir, xlogfile->name);
 
-		if (!get_wal_file(xlogfile->name, from_fullpath, to_fullpath, true))
+		if (!get_wal_file_wrapper(xlogfile->name, args->archive_dir, to_fullpath, true))
 		{
 			/* It is ok, maybe requested batch is greater than the number of available
 			 * files in the archive
@@ -1353,6 +1403,38 @@ get_files(void *arg)
 	return NULL;
 }
 
+/*
+ * First we try to copy from WAL archive subdirectory:
+ * Failing that, try WAL archive root directory
+ */
+bool
+get_wal_file_wrapper(const char *filename, const char *archive_root_dir,
+					 const char *to_fullpath, bool prefetch_mode)
+{
+	bool         success = false;
+	char         archive_subdir[MAXPGPATH];
+	char         from_fullpath[MAXPGPATH];
+	xlogFileType type = get_xlogFileType(filename);
+
+	if (type == SEGMENT || type == PARTIAL_SEGMENT || type == BACKUP_HISTORY_FILE)
+	{
+		/* first try subdir ... */
+		get_archive_subdir(archive_subdir, archive_root_dir, filename, type);
+		join_path_components(from_fullpath, archive_subdir, filename);
+
+		success = get_wal_file(filename, from_fullpath, to_fullpath, prefetch_mode);
+	}
+
+	if (!success)
+	{
+		/* ... fallback to archive dir for backward compatibility purposes */
+		join_path_components(from_fullpath, archive_root_dir, filename);
+		success = get_wal_file(filename, from_fullpath, to_fullpath, prefetch_mode);
+	}
+
+	return success;
+}
+
 /*
  * Copy WAL segment from archive catalog to pgdata with possible decompression.
  * When running in prefetch mode, we should not error out.
@@ -1755,3 +1837,30 @@ uint32 maintain_prefetch(const char *prefetch_dir, XLogSegNo first_segno, uint32
 
 	return n_files;
 }
+
+/* Calculate subdir path in WAL archive directory. Example:
+ * 000000010000000200000013 -> 00000002
+ */
+void
+get_archive_subdir(char *archive_subdir, const char *archive_dir, const char *wal_file_name, xlogFileType type)
+{
+	if (type == SEGMENT || type == PARTIAL_SEGMENT || type == BACKUP_HISTORY_FILE)
+	{
+		int  rc = 0;
+		char tli[MAXFNAMELEN];
+		char log[MAXFNAMELEN];
+		char suffix[MAXFNAMELEN];
+
+		rc = sscanf(wal_file_name, "%08s%08s%s",
+						(char *) &tli, (char *) &log, (char *) &suffix);
+
+		if (rc == 3)
+		{
+			join_path_components(archive_subdir, archive_dir, log);
+			return;
+		}
+	}
+
+	/* for all other files just use root directory of WAL archive */
+	strcpy(archive_subdir, archive_dir);
+}

src/backup.c

@@ -1238,6 +1238,7 @@ wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr target_lsn, bool is_start_l
 {
 	XLogSegNo	targetSegNo;
 	char		wal_segment_path[MAXPGPATH],
+				wal_segment_subdir[MAXPGPATH],
 				wal_segment[MAXFNAMELEN];
 	bool		file_exists = false;
 	uint32		try_count = 0,
@@ -1255,7 +1256,13 @@ wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr target_lsn, bool is_start_l
 	GetXLogFileName(wal_segment, tli, targetSegNo,
 					instance_config.xlog_seg_size);
 
-	join_path_components(wal_segment_path, wal_segment_dir, wal_segment);
+	// obtain WAL archive subdir for ARCHIVE backup
+	if (!in_stream_dir)
+		get_archive_subdir(wal_segment_subdir, wal_segment_dir, wal_segment, SEGMENT);
+	else
+		strcpy(wal_segment_subdir, wal_segment_dir);
+
+	join_path_components(wal_segment_path, wal_segment_subdir, wal_segment);
 	/*
 	 * In pg_start_backup we wait for 'target_lsn' in 'pg_wal' directory if it is
 	 * stream and non-page backup. Page backup needs archived WAL files, so we

src/catalog.c

@@ -1628,6 +1628,7 @@ catalog_get_timelines(InstanceState *instanceState, InstanceConfig *instance)
 				else if (strcmp(suffix, "gz") != 0)
 				{
 					elog(WARNING, "unexpected WAL file name \"%s\"", file->name);
+					// TODO: free file
 					continue;
 				}
 			}
@@ -1724,8 +1725,18 @@ catalog_get_timelines(InstanceState *instanceState, InstanceConfig *instance)
 			parray_walk(timelines, pfree);
 			parray_free(timelines);
 		}
+		/* add WAL archive subdirectories to filelist (used only in delete) */
+		else if (S_ISDIR(file->mode) && strspn(file->rel_path, "0123456789ABCDEF") == 8)
+		{
+			if (instanceState->wal_archive_subdirs == NULL)
+				instanceState->wal_archive_subdirs = parray_new();
+			parray_append(instanceState->wal_archive_subdirs, file);
+		}
 		else
+		{
 			elog(WARNING, "unexpected WAL file name \"%s\"", file->name);
+			// TODO: free file
+		}
 	}
 
 	/* save information about backups belonging to each timeline */

src/delete.c

@@ -932,14 +932,15 @@ delete_walfiles_in_tli(InstanceState *instanceState, XLogRecPtr keep_lsn, timeli
 		if (interrupted)
 			elog(ERROR, "interrupted during WAL archive purge");
 
-		/* Any segment equal or greater than EndSegNo must be kept
+		/*
+		 * Any segment equal or greater than EndSegNo must be kept
 		 * unless it`s a 'purge all' scenario.
 		 */
 		if (purge_all || wal_file->segno < OldestToKeepSegNo)
 		{
 			char wal_fullpath[MAXPGPATH];
 
-			join_path_components(wal_fullpath, instanceState->instance_wal_subdir_path, wal_file->file.name);
+			join_path_components(wal_fullpath, instanceState->instance_wal_subdir_path, wal_file->file.rel_path);
 
 			/* save segment from purging */
 			if (instance_config.wal_depth >= 0 && wal_file->keep)
@@ -970,6 +971,26 @@ delete_walfiles_in_tli(InstanceState *instanceState, XLogRecPtr keep_lsn, timeli
 
 			wal_deleted = true;
 		}
+
+		//TODO: cleanup
+	}
+
+	/* Remove empty subdirectories */
+	if (!instanceState->wal_archive_subdirs)
+		return;
+
+	for (i = 0; i < parray_num(instanceState->wal_archive_subdirs); i++)
+	{
+		char fullpath[MAXPGPATH];
+		pgFile *file = (pgFile *) parray_get(instanceState->wal_archive_subdirs, i);
+
+		join_path_components(fullpath, instanceState->instance_wal_subdir_path, file->name);
+
+		if (dir_is_empty(fullpath, FIO_LOCAL_HOST))
+			pgFileDelete(file->mode, fullpath);
+
+		pgFileFree(file);
+		// TODO: maintain instanceState->wal_archive_subdirs
 	}
 }
 

src/parsexlog.c

@@ -115,6 +115,7 @@ typedef struct XLogReaderData
 	gzFile		 gz_xlogfile;
 	char		 gz_xlogpath[MAXPGPATH];
 #endif
+	bool         is_stream;
 } XLogReaderData;
 
 /* Function to process a WAL record */
@@ -172,7 +173,8 @@ static bool RunXLogThreads(const char *archivedir,
 						   bool consistent_read,
 						   xlog_record_function process_record,
 						   XLogRecTarget *last_rec,
-						   bool inclusive_endpoint);
+						   bool inclusive_endpoint,
+						   bool is_stream);
 //static XLogReaderState *InitXLogThreadRead(xlog_thread_arg *arg);
 static bool SwitchThreadToNextWal(XLogReaderState *xlogreader,
 								  xlog_thread_arg *arg);
@@ -254,7 +256,7 @@ extractPageMap(const char *archivedir, uint32 wal_seg_size,
 		extract_isok = RunXLogThreads(archivedir, 0, InvalidTransactionId,
 									  InvalidXLogRecPtr, end_tli, wal_seg_size,
 									  startpoint, endpoint, false, extractPageInfo,
-									  NULL, true);
+									  NULL, true, false);
 	else
 	{
 		/* We have to process WAL located on several different xlog intervals,
@@ -348,7 +350,7 @@ extractPageMap(const char *archivedir, uint32 wal_seg_size,
 			extract_isok = RunXLogThreads(archivedir, 0, InvalidTransactionId,
 									  InvalidXLogRecPtr, tmp_interval->tli, wal_seg_size,
 									  tmp_interval->begin_lsn, tmp_interval->end_lsn,
-									  false, extractPageInfo, NULL, inclusive_endpoint);
+									  false, extractPageInfo, NULL, inclusive_endpoint, false);
 			if (!extract_isok)
 				break;
 
@@ -377,7 +379,7 @@ validate_backup_wal_from_start_to_stop(pgBackup *backup,
 	got_endpoint = RunXLogThreads(archivedir, 0, InvalidTransactionId,
 								  InvalidXLogRecPtr, tli, xlog_seg_size,
 								  backup->start_lsn, backup->stop_lsn,
-								  false, NULL, NULL, true);
+								  false, NULL, NULL, true, backup->stream);
 
 	if (!got_endpoint)
 	{
@@ -490,7 +492,8 @@ validate_wal(pgBackup *backup, const char *archivedir,
 	all_wal = all_wal ||
 		RunXLogThreads(archivedir, target_time, target_xid, target_lsn,
 					   tli, wal_seg_size, backup->stop_lsn,
-					   InvalidXLogRecPtr, true, validateXLogRecord, &last_rec, true);
+					   InvalidXLogRecPtr, true, validateXLogRecord, &last_rec, true,
+					   backup->stream);
 	if (last_rec.rec_time > 0)
 		time2iso(last_timestamp, lengthof(last_timestamp),
 				 timestamptz_to_time_t(last_rec.rec_time), false);
@@ -1017,7 +1020,16 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 
 		GetXLogFileName(xlogfname, reader_data->tli, reader_data->xlogsegno, wal_seg_size);
 
-		join_path_components(reader_data->xlogpath, wal_archivedir, xlogfname);
+		if (reader_data->is_stream)
+			join_path_components(reader_data->xlogpath, wal_archivedir, xlogfname);
+		/* obtain WAL archive subdir for ARCHIVE backup */
+		else
+		{
+			char	archive_subdir[MAXPGPATH];
+			get_archive_subdir(archive_subdir, wal_archivedir, xlogfname, SEGMENT);
+			join_path_components(reader_data->xlogpath, archive_subdir, xlogfname);
+		}
+
 		snprintf(reader_data->gz_xlogpath, MAXPGPATH, "%s.gz", reader_data->xlogpath);
 
 		/* We fall back to using .partial segment in case if we are running
@@ -1191,7 +1203,7 @@ RunXLogThreads(const char *archivedir, time_t target_time,
 			   TransactionId target_xid, XLogRecPtr target_lsn, TimeLineID tli,
 			   uint32 segment_size, XLogRecPtr startpoint, XLogRecPtr endpoint,
 			   bool consistent_read, xlog_record_function process_record,
-			   XLogRecTarget *last_rec, bool inclusive_endpoint)
+			   XLogRecTarget *last_rec, bool inclusive_endpoint, bool is_stream)
 {
 	pthread_t  *threads;
 	xlog_thread_arg *thread_args;
@@ -1255,6 +1267,7 @@ RunXLogThreads(const char *archivedir, time_t target_time,
 						 consistent_read, false);
 		arg->reader_data.xlogsegno = segno_next;
 		arg->reader_data.thread_num = i + 1;
+		arg->reader_data.is_stream = is_stream;
 		arg->process_record = process_record;
 		arg->startpoint = startpoint;
 		arg->endpoint = endpoint;
@@ -1915,7 +1928,7 @@ bool validate_wal_segment(TimeLineID tli, XLogSegNo segno, const char *prefetch_
 
 	rc = RunXLogThreads(prefetch_dir, 0, InvalidTransactionId,
 						InvalidXLogRecPtr, tli, wal_seg_size,
-						startpoint, endpoint, false, NULL, NULL, true);
+						startpoint, endpoint, false, NULL, NULL, true, true);
 
 	num_threads = tmp_num_threads;
 

src/pg_probackup.c

@@ -494,6 +494,7 @@ main(int argc, char *argv[])
 							catalogState->wal_subdir_path, instanceState->instance_name);
 		join_path_components(instanceState->instance_config_path,
 							 instanceState->instance_backup_subdir_path, BACKUP_CATALOG_CONF_FILE);
+		instanceState->wal_archive_subdirs = NULL;
 
 	}
 	/* ===== instanceState (END) ======*/

src/pg_probackup.h

@@ -647,9 +647,11 @@ typedef struct lsnInterval
 
 typedef enum xlogFileType
 {
+	UNKNOWN,
 	SEGMENT,
-	TEMP_SEGMENT,
+	TEMP_SEGMENT, // '.part' segment created by archive-push 
 	PARTIAL_SEGMENT,
+	HISTORY_FILE,
 	BACKUP_HISTORY_FILE
 } xlogFileType;
 
@@ -814,6 +816,8 @@ typedef struct InstanceState
 	/* $BACKUP_PATH/backups/instance_name */
 	char		instance_wal_subdir_path[MAXPGPATH]; // previously global var arclog_path
 
+	parray     *wal_archive_subdirs;
+
 	/* TODO: Make it more specific */
 	PGconn *conn;
 
@@ -894,6 +898,7 @@ extern void do_archive_push(InstanceState *instanceState, InstanceConfig *instan
 						   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,
 						   char *wal_file_name, int batch_size, bool validate_wal);
+extern void get_archive_subdir(char *archive_subdir, const char * archive_dir, const char *wal_file_name, xlogFileType type);
 
 /* in configure.c */
 extern void do_show_config(void);