Merge branch 'pgpro_428_v1'

Author: za-arthur

src/catalog.c

@@ -254,6 +254,7 @@ catalog_get_backup_list(time_t requested_backup_id)
 	struct dirent  *data_ent = NULL;
 	parray		   *backups = NULL;
 	pgBackup	   *backup = NULL;
+	int		i;
 
 	/* open backup instance backups directory */
 	data_dir = opendir(backup_instance_path);
@@ -282,6 +283,7 @@ catalog_get_backup_list(time_t requested_backup_id)
 		/* read backup information from BACKUP_CONTROL_FILE */
 		snprintf(backup_conf_path, MAXPGPATH, "%s/%s", data_path, BACKUP_CONTROL_FILE);
 		backup = readBackupControlFile(backup_conf_path);
+		backup->backup_id = backup->start_time;
 
 		/* ignore corrupted backups */
 		if (backup)
@@ -315,6 +317,30 @@ catalog_get_backup_list(time_t requested_backup_id)
 
 	parray_qsort(backups, pgBackupCompareIdDesc);
 
+	/* Link incremental backups with their ancestors.*/
+	for (i = 0; i < parray_num(backups); i++)
+	{
+		pgBackup *curr = parray_get(backups, i);
+
+		int j;
+
+		if (curr->backup_mode == BACKUP_MODE_FULL)
+			continue;
+
+		for (j = i+1; j < parray_num(backups); j++)
+		{
+			pgBackup *ancestor = parray_get(backups, j);
+
+			if (ancestor->start_time == curr->parent_backup)
+			{
+				curr->parent_backup_link = ancestor;
+				/* elog(INFO, "curr %s, ancestor %s j=%d", base36enc_dup(curr->start_time),
+						base36enc_dup(ancestor->start_time), j); */
+				break;
+			}
+		}
+	}
+
 	return backups;
 
 err_proc:
@@ -757,6 +783,7 @@ pgBackupInit(pgBackup *backup)
 	backup->stream = false;
 	backup->from_replica = false;
 	backup->parent_backup = INVALID_BACKUP_ID;
+	backup->parent_backup_link = NULL;
 	backup->primary_conninfo = NULL;
 	backup->program_version[0] = '\0';
 	backup->server_version[0] = '\0';
@@ -840,3 +867,48 @@ pgBackupGetPath2(const pgBackup *backup, char *path, size_t len,
 
 	make_native_path(path);
 }
+
+/* Find parent base FULL backup for current backup using parent_backup_link,
+ * return NULL if not found
+ */
+pgBackup*
+find_parent_backup(pgBackup *current_backup)
+{
+	pgBackup   *base_full_backup = NULL;
+	base_full_backup = current_backup;
+
+	while (base_full_backup->backup_mode != BACKUP_MODE_FULL)
+	{
+		/*
+		 * If we haven't found parent for incremental backup,
+		 * mark it and all depending backups as orphaned
+		 */
+		if (base_full_backup->parent_backup_link == NULL
+			|| (base_full_backup->status != BACKUP_STATUS_OK
+				&& base_full_backup->status != BACKUP_STATUS_DONE))
+		{
+			pgBackup   *orphaned_backup = current_backup;
+
+			while (orphaned_backup != NULL)
+			{
+				orphaned_backup->status = BACKUP_STATUS_ORPHAN;
+				pgBackupWriteBackupControlFile(orphaned_backup);
+				if (base_full_backup->parent_backup_link == NULL)
+					elog(WARNING, "Backup %s is orphaned because its parent backup is not found",
+							base36enc(orphaned_backup->start_time));
+				else
+					elog(WARNING, "Backup %s is orphaned because its parent backup is corrupted",
+							base36enc(orphaned_backup->start_time));
+
+				orphaned_backup = orphaned_backup->parent_backup_link;
+			}
+
+			base_full_backup = NULL;
+			break;
+		}
+
+		base_full_backup = base_full_backup->parent_backup_link;
+	}
+
+	return base_full_backup;
+}

src/pg_probackup.h

@@ -203,6 +203,8 @@ typedef struct pgBackupConfig
 	int			compress_level;
 } pgBackupConfig;
 
+typedef struct pgBackup pgBackup;
+
 /* Information about single backup stored in backup.conf */
 typedef struct pgBackup
 {
@@ -250,8 +252,9 @@ typedef struct pgBackup
 	time_t			parent_backup; 	/* Identifier of the previous backup.
 									 * Which is basic backup for this
 									 * incremental backup. */
-	char		   *primary_conninfo; /* Connection parameters of the backup
-									   * in the format suitable for recovery.conf */
+	pgBackup		*parent_backup_link;
+	char			*primary_conninfo; /* Connection parameters of the backup
+										* in the format suitable for recovery.conf */
 } pgBackup;
 
 /* Recovery target for restore and validate subcommands */
@@ -486,6 +489,8 @@ extern void pgBackupFree(void *backup);
 extern int pgBackupCompareId(const void *f1, const void *f2);
 extern int pgBackupCompareIdDesc(const void *f1, const void *f2);
 
+extern pgBackup* find_parent_backup(pgBackup *current_backup);
+
 /* in dir.c */
 extern void dir_list_file(parray *files, const char *root, bool exclude,
 						  bool omit_symlink, bool add_root);

src/restore.c

@@ -47,7 +47,7 @@ int
 do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 					   bool is_restore)
 {
-	int			i;
+	int			i = 0;
 	parray	   *backups;
 	pgBackup   *current_backup = NULL;
 	pgBackup   *dest_backup = NULL;
@@ -79,9 +79,10 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 	backups = catalog_get_backup_list(INVALID_BACKUP_ID);
 
 	/* Find backup range we should restore or validate. */
-	for (i = 0; i < parray_num(backups); i++)
+	while ((i < parray_num(backups)) && !dest_backup)
 	{
 		current_backup = (pgBackup *) parray_get(backups, i);
+		i++;
 
 		/* Skip all backups which started after target backup */
 		if (target_backup_id && current_backup->start_time > target_backup_id)
@@ -93,7 +94,6 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 		 */
 
 		if (is_restore &&
-			!dest_backup &&
 			target_backup_id == INVALID_BACKUP_ID &&
 			current_backup->status != BACKUP_STATUS_OK)
 		{
@@ -107,8 +107,7 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 		 * ensure that it satisfies recovery target.
 		 */
 		if ((target_backup_id == current_backup->start_time
-			|| target_backup_id == INVALID_BACKUP_ID)
-			&& !dest_backup)
+			|| target_backup_id == INVALID_BACKUP_ID))
 		{
 
 			/* backup is not ok,
@@ -161,37 +160,42 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 			 * Save it as dest_backup
 			 */
 			dest_backup = current_backup;
-			dest_backup_index = i;
+			dest_backup_index = i-1;
+		}
+	}
+
+	if (dest_backup == NULL)
+		elog(ERROR, "Backup satisfying target options is not found.");
+
+	/* If we already found dest_backup, look for full backup. */
+	if (dest_backup)
+	{
+		base_full_backup = current_backup;
+
+		if (current_backup->backup_mode != BACKUP_MODE_FULL)
+		{
+			base_full_backup = find_parent_backup(current_backup);
+
+			if (base_full_backup == NULL)
+				elog(ERROR, "Valid full backup for backup %s is not found.",
+					base36enc(current_backup->start_time));
 		}
 
-		/* If we already found dest_backup, look for full backup. */
-		if (dest_backup)
+		/*
+		 * We have found full backup by link,
+		 * now we need to walk the list to find its index.
+		 *
+		 * TODO I think we should rewrite it someday to use double linked list
+		 * and avoid relying on sort order anymore.
+		 */
+		for (i = dest_backup_index; i < parray_num(backups); i++)
 		{
-			if (current_backup->backup_mode == BACKUP_MODE_FULL)
+			pgBackup * temp_backup = (pgBackup *) parray_get(backups, i);
+			if (temp_backup->start_time == base_full_backup->start_time)
 			{
-				if (current_backup->status != BACKUP_STATUS_OK)
-				{
-					/* Full backup revalidation can be done only for DONE and CORRUPT */
-					if (current_backup->status == BACKUP_STATUS_DONE ||
-							current_backup->status == BACKUP_STATUS_CORRUPT)
-						elog(WARNING, "base backup %s for given backup %s is in %s status, trying to revalidate",
-							base36enc_dup(current_backup->start_time),
-							base36enc_dup(dest_backup->start_time),
-							status2str(current_backup->status));
-					else
-						elog(ERROR, "base backup %s for given backup %s is in %s status",
-							base36enc_dup(current_backup->start_time),
-							base36enc_dup(dest_backup->start_time),
-							status2str(current_backup->status));
-				}
-				/* We found both dest and base backups. */
-				base_full_backup = current_backup;
 				base_full_backup_index = i;
 				break;
 			}
-			else
-				/* It`s ok to skip incremental backup */
-				continue;
 		}
 	}
 

src/validate.c

@@ -291,43 +291,33 @@ do_validate_instance(void)
 	/* Get list of all backups sorted in order of descending start time */
 	backups = catalog_get_backup_list(INVALID_BACKUP_ID);
 
-	/* Valiate each backup along with its xlog files. */
+	/* Examine backups one by one and validate them */
 	for (i = 0; i < parray_num(backups); i++)
 	{
-		pgBackup   *base_full_backup = NULL;
-
 		current_backup = (pgBackup *) parray_get(backups, i);
 
-		if (current_backup->backup_mode != BACKUP_MODE_FULL)
+		/* Valiate each backup along with its xlog files. */
+		pgBackupValidate(current_backup);
+
+		/* Ensure that the backup has valid list of parent backups */
+		if (current_backup->status == BACKUP_STATUS_OK)
 		{
-			int			j;
+			pgBackup   *base_full_backup = current_backup;
 
-			for (j = i + 1; j < parray_num(backups); j++)
+			if (current_backup->backup_mode != BACKUP_MODE_FULL)
 			{
-				pgBackup	   *backup = (pgBackup *) parray_get(backups, j);
+				base_full_backup = find_parent_backup(current_backup);
 
-				if (backup->backup_mode == BACKUP_MODE_FULL)
-				{
-					base_full_backup = backup;
-					break;
-				}
+				if (base_full_backup == NULL)
+					elog(ERROR, "Valid full backup for backup %s is not found.",
+						 base36enc(current_backup->start_time));
 			}
-		}
-		else
-			base_full_backup = current_backup;
-
-		pgBackupValidate(current_backup);
 
-		/* There is no point in wal validation for corrupted backup */
-		if (current_backup->status == BACKUP_STATUS_OK)
-		{
-			if (base_full_backup == NULL)
-				elog(ERROR, "Valid full backup for backup %s is not found.",
-					 base36enc(current_backup->start_time));
 			/* Validate corresponding WAL files */
 			validate_wal(current_backup, arclog_path, 0,
-						0, 0, base_full_backup->tli);
+						 0, 0, base_full_backup->tli);
 		}
+
 		/* Mark every incremental backup between corrupted backup and nearest FULL backup as orphans */
 		if (current_backup->status == BACKUP_STATUS_CORRUPT)
 		{