Merge branch 'master' into fix_ptrack_1230

Author: alubennikova

src/backup.c

@@ -1541,23 +1541,18 @@ pg_stop_backup(pgBackup *backup)
 	{
 		const char *params[1];
 		char		name[1024];
-		char	   *backup_id;
-
-		backup_id = base36enc(backup->start_time);
 
 		if (!from_replica)
 			snprintf(name, lengthof(name), "pg_probackup, backup_id %s",
-					 backup_id);
+					 base36enc(backup->start_time));
 		else
 			snprintf(name, lengthof(name), "pg_probackup, backup_id %s. Replica Backup",
-					 backup_id);
+					 base36enc(backup->start_time));
 		params[0] = name;
 
 		res = pgut_execute(conn, "SELECT pg_create_restore_point($1)",
 						   1, params, true);
 		PQclear(res);
-
-		pfree(backup_id);
 	}
 
 	/*
@@ -1845,7 +1840,8 @@ backup_cleanup(bool fatal, void *userdata)
 	 */
 	if (current.status == BACKUP_STATUS_RUNNING && current.end_time == 0)
 	{
-		elog(INFO, "Backup %s is running, setting its status to ERROR", base36enc(current.start_time));
+		elog(INFO, "Backup %s is running, setting its status to ERROR",
+			 base36enc(current.start_time));
 		current.end_time = time(NULL);
 		current.status = BACKUP_STATUS_ERROR;
 		pgBackupWriteBackupControlFile(&current);

src/catalog.c

@@ -436,12 +436,7 @@ pgBackupWriteControl(FILE *out, pgBackup *backup)
 
 	/* 'parent_backup' is set if it is incremental backup */
 	if (backup->parent_backup != 0)
-	{
-		char	   *parent_backup = base36enc(backup->parent_backup);
-
-		fprintf(out, "parent-backup-id = '%s'\n", parent_backup);
-		free(parent_backup);
-	}
+		fprintf(out, "parent-backup-id = '%s'\n", base36enc(backup->parent_backup));
 }
 
 /* create BACKUP_CONTROL_FILE */
@@ -556,6 +551,8 @@ readBackupControlFile(const char *path)
 			backup->status = BACKUP_STATUS_DELETED;
 		else if (strcmp(status, "DONE") == 0)
 			backup->status = BACKUP_STATUS_DONE;
+		else if (strcmp(status, "ORPHAN") == 0)
+			backup->status = BACKUP_STATUS_ORPHAN;
 		else if (strcmp(status, "CORRUPT") == 0)
 			backup->status = BACKUP_STATUS_CORRUPT;
 		else
@@ -667,21 +664,17 @@ void
 pgBackupGetPath2(const pgBackup *backup, char *path, size_t len,
 				 const char *subdir1, const char *subdir2)
 {
-	char	   *datetime;
-
-	datetime = base36enc(backup->start_time);
-
 	/* If "subdir1" is NULL do not check "subdir2" */
 	if (!subdir1)
-		snprintf(path, len, "%s/%s", backup_instance_path, datetime);
+		snprintf(path, len, "%s/%s", backup_instance_path,
+				 base36enc(backup->start_time));
 	else if (!subdir2)
-		snprintf(path, len, "%s/%s/%s", backup_instance_path, datetime, subdir1);
+		snprintf(path, len, "%s/%s/%s", backup_instance_path,
+				 base36enc(backup->start_time), subdir1);
 	/* "subdir1" and "subdir2" is not NULL */
 	else
 		snprintf(path, len, "%s/%s/%s/%s", backup_instance_path,
-				 datetime, subdir1, subdir2);
-
-	free(datetime);
+				 base36enc(backup->start_time), subdir1, subdir2);
 
 	make_native_path(path);
 }

src/delete.c

@@ -224,7 +224,6 @@ static int
 pgBackupDeleteFiles(pgBackup *backup)
 {
 	size_t		i;
-	char	   *backup_id;
 	char		path[MAXPGPATH];
 	char		timestamp[100];
 	parray	   *files;
@@ -235,11 +234,9 @@ pgBackupDeleteFiles(pgBackup *backup)
 	if (backup->status == BACKUP_STATUS_DELETED)
 		return 0;
 
-	backup_id = base36enc(backup->start_time);
 	time2iso(timestamp, lengthof(timestamp), backup->recovery_time);
 
-	elog(INFO, "delete: %s %s", backup_id, timestamp);
-	free(backup_id);
+	elog(INFO, "delete: %s %s", base36enc(backup->start_time), timestamp);
 
 	/*
 	 * Update STATUS to BACKUP_STATUS_DELETING in preparation for the case which

src/parsexlog.c

@@ -258,7 +258,9 @@ validate_backup_wal_from_start_to_stop(pgBackup *backup,
 		 */
 			backup->status = BACKUP_STATUS_CORRUPT;
 			pgBackupWriteBackupControlFile(backup);
-			elog(ERROR, "there are not enough WAL records to restore from %X/%X to %X/%X",
+			elog(WARNING, "There are not enough WAL records to consistenly restore "
+				"backup %s from START LSN: %X/%X to STOP LSN: %X/%X",
+				 base36enc(backup->start_time),
 				 (uint32) (backup->start_lsn >> 32),
 				 (uint32) (backup->start_lsn),
 				 (uint32) (backup->stop_lsn >> 32),
@@ -288,7 +290,7 @@ validate_wal(pgBackup *backup,
 			 TimeLineID tli)
 {
 	XLogRecPtr	startpoint = backup->start_lsn;
-	char	   *backup_id;
+	const char *backup_id;
 	XLogRecord *record;
 	XLogReaderState *xlogreader;
 	char	   *errormsg;
@@ -327,16 +329,19 @@ validate_wal(pgBackup *backup,
 	else
 		validate_backup_wal_from_start_to_stop(backup, (char *) archivedir, tli);
 
-	free(backup_id);
-
+	if (backup->status == BACKUP_STATUS_CORRUPT)
+	{
+		elog(WARNING, "Backup %s WAL segments are corrupted", backup_id);
+		return;
+	}
 	/*
 	 * If recovery target is provided check that we can restore backup to a
-	 * recoverty target time or xid.
+	 * recovery target time or xid.
 	 */
 	if (!TransactionIdIsValid(target_xid) && target_time == 0)
 	{
-		/* Recoverty target is not given so exit */
-		elog(INFO, "backup validation completed successfully");
+		/* Recovery target is not given so exit */
+		elog(INFO, "Backup %s WAL segments are valid", backup_id);
 		return;
 	}
 

src/pg_probackup.c

@@ -423,21 +423,17 @@ main(int argc, char *argv[])
 			return do_init();
 		case BACKUP:
 			{
-				char	   *backup_id;
 				const char *backup_mode;
 				time_t		start_time;
 
 				start_time = time(NULL);
-				backup_id = base36enc(start_time);
 				backup_mode = deparse_backup_mode(current.backup_mode);
 
 				elog(INFO, "Backup start, pg_probackup version: %s, backup ID: %s, backup mode: %s, instance: %s, stream: %s, remote: %s",
-						  PROGRAM_VERSION, backup_id, backup_mode, instance_name,
+						  PROGRAM_VERSION, base36enc(start_time), backup_mode, instance_name,
 						  current.stream ? "true" : "false", is_remote_backup ? "true" : "false");
 				elog_file(INFO, "command: %s", command);
 
-				pfree(backup_id);
-
 				return do_backup(start_time);
 			}
 		case RESTORE:

src/pg_probackup.h

@@ -118,6 +118,7 @@ typedef enum BackupStatus
 	BACKUP_STATUS_DELETING,		/* data files are being deleted */
 	BACKUP_STATUS_DELETED,		/* data files have been deleted */
 	BACKUP_STATUS_DONE,			/* completed but not validated yet */
+	BACKUP_STATUS_ORPHAN,		/* backup validity is unknown but at least one parent backup is corrupted */
 	BACKUP_STATUS_CORRUPT		/* files are corrupted, not available */
 } BackupStatus;
 
@@ -463,7 +464,8 @@ extern const char *status2str(BackupStatus status);
 extern void remove_trailing_space(char *buf, int comment_mark);
 extern void remove_not_digit(char *buf, size_t len, const char *str);
 extern uint32 get_data_checksum_version(bool safe);
-extern char *base36enc(long unsigned int value);
+extern const char *base36enc(long unsigned int value);
+extern char *base36enc_dup(long unsigned int value);
 extern long unsigned int base36dec(const char *text);
 extern uint64 get_system_identifier(char *pgdata);
 extern uint64 get_remote_system_identifier(PGconn *conn);

src/restore.c

@@ -90,8 +90,10 @@ do_restore_or_validate(time_t target_backup_id,
 	pgBackup   *current_backup = NULL;
 	pgBackup   *dest_backup = NULL;
 	pgBackup   *base_full_backup = NULL;
+	pgBackup   *corrupted_backup = NULL;
 	int			dest_backup_index = 0;
 	int			base_full_backup_index = 0;
+	int			corrupted_backup_index = 0;
 	char 	   *action = is_restore ? "Restore":"Validate";
 
 	if (is_restore)
@@ -193,8 +195,8 @@ do_restore_or_validate(time_t target_backup_id,
 			{
 				if (current_backup->status != BACKUP_STATUS_OK)
 					elog(ERROR, "base backup %s for given backup %s is in %s status",
-						 base36enc(current_backup->start_time),
-						 base36enc(dest_backup->start_time),
+						 base36enc_dup(current_backup->start_time),
+						 base36enc_dup(dest_backup->start_time),
 						 status2str(current_backup->status));
 				else
 				{
@@ -205,7 +207,7 @@ do_restore_or_validate(time_t target_backup_id,
 				}
 			}
 			else
-				/* Skip differential backups are ok */
+				/* It`s ok to skip incremental backup */
 				continue;
 		}
 	}
@@ -220,36 +222,87 @@ do_restore_or_validate(time_t target_backup_id,
 	if (is_restore)
 		check_tablespace_mapping(dest_backup);
 
+	if (dest_backup->backup_mode != BACKUP_MODE_FULL)
+		elog(INFO, "Validating parents for backup %s", base36enc(dest_backup->start_time));
+
 	/*
 	 * Validate backups from base_full_backup to dest_backup.
 	 */
 	for (i = base_full_backup_index; i >= dest_backup_index; i--)
 	{
 		pgBackup   *backup = (pgBackup *) parray_get(backups, i);
 		pgBackupValidate(backup);
+		if (backup->status == BACKUP_STATUS_CORRUPT)
+		{
+			corrupted_backup = backup;
+			corrupted_backup_index = i;
+			break;
+		}
+	}
+	/* There is no point in wal validation
+	 * if there is corrupted backup between base_backup and dest_backup
+	 */
+	if (!corrupted_backup)
+		/*
+		 * Validate corresponding WAL files.
+		 * We pass base_full_backup timeline as last argument to this function,
+		 * because it's needed to form the name of xlog file.
+		 */
+		validate_wal(dest_backup, arclog_path, rt->recovery_target_time,
+					rt->recovery_target_xid, base_full_backup->tli);
+
+	/* Set every incremental backup between corrupted backup and nearest FULL backup as orphans */
+	if (corrupted_backup)
+	{
+		for (i = corrupted_backup_index - 1; i >= 0; i--)
+		{
+			pgBackup   *backup = (pgBackup *) parray_get(backups, i);
+			/* Mark incremental OK backup as orphan */
+			if (backup->backup_mode == BACKUP_MODE_FULL)
+				break;
+			if (backup->status != BACKUP_STATUS_OK)
+				continue;
+			else
+			{
+				char	   *backup_id,
+						   *corrupted_backup_id;
+
+				backup->status = BACKUP_STATUS_ORPHAN;
+				pgBackupWriteBackupControlFile(backup);
+
+				backup_id = base36enc_dup(backup->start_time);
+				corrupted_backup_id = base36enc_dup(corrupted_backup->start_time);
+
+				elog(WARNING, "Backup %s is orphaned because his parent %s is corrupted",
+					 backup_id, corrupted_backup_id);
+
+				free(backup_id);
+				free(corrupted_backup_id);
+			}
+		}
 	}
 
 	/*
-	 * Validate corresponding WAL files.
-	 * We pass base_full_backup timeline as last argument to this function,
-	 * because it's needed to form the name of xlog file.
+	 * If dest backup is corrupted or was orphaned in previous check
+	 * produce corresponding error message
 	 */
-	validate_wal(dest_backup, arclog_path, rt->recovery_target_time,
-						 rt->recovery_target_xid, base_full_backup->tli);
-
+	if (dest_backup->status == BACKUP_STATUS_OK)
+		elog(INFO, "Backup %s is valid.", base36enc(dest_backup->start_time));
+	else if (dest_backup->status == BACKUP_STATUS_CORRUPT)
+		elog(ERROR, "Backup %s is corrupt.", base36enc(dest_backup->start_time));
+	else if (dest_backup->status == BACKUP_STATUS_ORPHAN)
+		elog(ERROR, "Backup %s is orphan.", base36enc(dest_backup->start_time));
+	else
+		elog(ERROR, "Backup %s has status: %s",
+				base36enc(dest_backup->start_time), status2str(dest_backup->status));
 
 	/* We ensured that all backups are valid, now restore if required */
 	if (is_restore)
 	{
-		pgBackup   *backup;
 		for (i = base_full_backup_index; i >= dest_backup_index; i--)
 		{
-			backup = (pgBackup *) parray_get(backups, i);
-			if (backup->status == BACKUP_STATUS_OK)
-				restore_backup(backup);
-			else
-				elog(ERROR, "backup %s is not valid",
-					 base36enc(backup->start_time));
+			pgBackup   *backup = (pgBackup *) parray_get(backups, i);
+			restore_backup(backup);
 		}
 
 		/*
@@ -366,13 +419,7 @@ restore_backup(pgBackup *backup)
 	parray_free(files);
 
 	if (LOG_LEVEL_CONSOLE <= LOG || LOG_LEVEL_FILE <= LOG)
-	{
-		char	   *backup_id;
-
-		backup_id = base36enc(backup->start_time);
-		elog(LOG, "restore %s backup completed", backup_id);
-		free(backup_id);
-	}
+		elog(LOG, "restore %s backup completed", base36enc(backup->start_time));
 }
 
 /*
@@ -592,13 +639,8 @@ check_tablespace_mapping(pgBackup *backup)
 	read_tablespace_map(links, this_backup_path);
 
 	if (LOG_LEVEL_CONSOLE <= LOG || LOG_LEVEL_FILE <= LOG)
-	{
-		char	   *backup_id;
-
-		backup_id = base36enc(backup->start_time);
-		elog(LOG, "check tablespace directories of backup %s", backup_id);
-		pfree(backup_id);
-	}
+		elog(LOG, "check tablespace directories of backup %s",
+			 base36enc(backup->start_time));
 
 	/* 1 - each OLDDIR must have an entry in tablespace_map file (links) */
 	for (cell = tablespace_dirs.head; cell; cell = cell->next)

src/show.c

@@ -234,7 +234,6 @@ show_backup_list(FILE *out, parray *backup_list)
 	{
 		pgBackup   *backup = parray_get(backup_list, i);
 		TimeLineID	parent_tli;
-		char	   *backup_id;
 		char		timestamp[100] = "----";
 		char		duration[20] = "----";
 		char		data_bytes_str[10] = "----";
@@ -255,12 +254,11 @@ show_backup_list(FILE *out, parray *backup_list)
 
 		/* Get parent timeline before printing */
 		parent_tli = get_parent_tli(backup->tli);
-		backup_id = base36enc(backup->start_time);
 
 		fprintf(out, " %-11s %-8s %-6s  %-22s  %-6s  %-7s  %3d / %-3d            %5s  %6s  %2X/%-8X  %2X/%-8X  %-8s\n",
 				instance_name,
 				(backup->server_version[0] ? backup->server_version : "----"),
-				backup_id,
+				base36enc(backup->start_time),
 				timestamp,
 				pgBackupGetBackupMode(backup),
 				backup->stream ? "STREAM": "ARCHIVE",
@@ -273,8 +271,6 @@ show_backup_list(FILE *out, parray *backup_list)
 				(uint32) (backup->stop_lsn >> 32),
 				(uint32) backup->stop_lsn,
 				status2str(backup->status));
-
-		free(backup_id);
 	}
 }