Implement --report-dir option for gprestore (#30)

Implement --report-dir option for gprestore

New option allows to create report files in directory different from --backup-dir.
Main code changes were made for filepath class. Now it takes new option as deafult path for reports with fallback to content dir.
Unit and end_to_end tests were updated to show mentioned behavior.

Cherry-picked-from: ac8731e
to reapply above 8bdcb48 and 7346404

end_to_end/end_to_end_suite_test.go

@@ -363,6 +363,18 @@ func getMetdataFileContents(backupDir string, timestamp string, fileSuffix strin
 	return fileContentBytes
 }
 
+// check restore file exist and has right permissions
+func checkRestoreMetdataFile(backupDir string, timestamp string, fileSuffix string) {
+	file, err := path.Glob(path.Join(backupDir, "*-1/backups", timestamp[:8], timestamp, fmt.Sprintf("gprestore_%s_*_%s", timestamp, fileSuffix)))
+	Expect(err).ToNot(HaveOccurred())
+	Expect(file).To(HaveLen(1))
+	info, err := os.Stat(file[0])
+	Expect(err).ToNot(HaveOccurred())
+	if info.Mode() != 0444 {
+		Fail(fmt.Sprintf("File %s is not read-only (mode is %v).", file[0], info.Mode()))
+	}
+}
+
 func saveHistory(myCluster *cluster.Cluster) {
 	// move history file out of the way, and replace in "after". This avoids adding junk to an existing gpackup_history.db
 
@@ -1202,6 +1214,75 @@ var _ = Describe("backup and restore end to end tests", func() {
 			Expect(actualStatisticCount).To(Equal("3"))
 		})
 	})
+	Describe("Restore with --report-dir", func() {
+		It("runs gprestore without --report-dir", func() {
+			timestamp := gpbackup(gpbackupPath, backupHelperPath,
+				"--include-table", "public.sales")
+			gprestore(gprestorePath, restoreHelperPath, timestamp,
+				"--redirect-db", "restoredb")
+
+			// Since --report-dir and --backup-dir were not used, restore report should be in default dir
+			checkRestoreMetdataFile(path.Dir(backupCluster.GetDirForContent(-1)), timestamp, "report")
+		})
+		It("runs gprestore without --report-dir, but with --backup-dir", func() {
+			timestamp := gpbackup(gpbackupPath, backupHelperPath,
+				"--backup-dir", backupDir,
+				"--include-table", "public.sales")
+			gprestore(gprestorePath, restoreHelperPath, timestamp,
+				"--backup-dir", backupDir,
+				"--redirect-db", "restoredb")
+
+			// Since --backup-dir was used, restore report should be in backup dir
+			checkRestoreMetdataFile(backupDir, timestamp, "report")
+		})
+		It("runs gprestore with --report-dir and same --backup-dir", func() {
+			timestamp := gpbackup(gpbackupPath, backupHelperPath,
+				"--backup-dir", backupDir,
+				"--include-table", "public.sales")
+			gprestore(gprestorePath, restoreHelperPath, timestamp,
+				"--backup-dir", backupDir,
+				"--report-dir", backupDir,
+				"--redirect-db", "restoredb")
+
+			// Since --report-dir and --backup-dir are the same, restore report should be in backup dir
+			checkRestoreMetdataFile(backupDir, timestamp, "report")
+		})
+		It("runs gprestore with --report-dir and different --backup-dir", func() {
+			reportDir := path.Join(backupDir, "restore")
+			timestamp := gpbackup(gpbackupPath, backupHelperPath,
+				"--backup-dir", backupDir,
+				"--include-table", "public.sales")
+			gprestore(gprestorePath, restoreHelperPath, timestamp,
+				"--backup-dir", backupDir,
+				"--report-dir", reportDir,
+				"--redirect-db", "restoredb")
+
+			// Since --report-dir differs from --backup-dir, restore report should be in report dir
+			checkRestoreMetdataFile(reportDir, timestamp, "report")
+		})
+		It("runs gprestore with --report-dir and check error_tables* files are present", func() {
+			if segmentCount != 3 {
+				Skip("Restoring from a tarred backup currently requires a 3-segment cluster to test.")
+			}
+			extractDirectory := extractSavedTarFile(backupDir, "corrupt-db")
+			reportDir := path.Join(backupDir, "restore")
+
+			// Restore command with data error
+			// Metadata errors due to invalid alter ownership
+			gprestoreCmd := exec.Command(gprestorePath,
+				"--timestamp", "20190809230424",
+				"--redirect-db", "restoredb",
+				"--backup-dir", extractDirectory,
+				"--report-dir", reportDir,
+				"--on-error-continue")
+			_, _ = gprestoreCmd.CombinedOutput()
+
+			// All report files should be placed in the same dir
+			checkRestoreMetdataFile(reportDir, "20190809230424", "report")
+			checkRestoreMetdataFile(reportDir, "20190809230424", "error_tables_metadata")
+			checkRestoreMetdataFile(reportDir, "20190809230424", "error_tables_data")
+		})
+	})
 	Describe("Flag combinations", func() {
 		It("runs gpbackup and gprestore without redirecting restore to another db", func() {
 			err := exec.Command("createdb", "recreateme").Run()

filepath/filepath.go

@@ -27,13 +27,15 @@ type FilePathInfo struct {
 	UserSpecifiedBackupDir string
 	UserSpecifiedSegPrefix string
 	BaseDataDir            string
+	UserSpecifiedReportDir string
 }
 
 func NewFilePathInfo(c *cluster.Cluster, userSpecifiedBackupDir string, timestamp string, userSegPrefix string, useMirrors ...bool) FilePathInfo {
 	backupFPInfo := FilePathInfo{}
 	backupFPInfo.PID = os.Getpid()
 	backupFPInfo.UserSpecifiedBackupDir = userSpecifiedBackupDir
 	backupFPInfo.UserSpecifiedSegPrefix = userSegPrefix
+	backupFPInfo.UserSpecifiedReportDir = ""
 	backupFPInfo.Timestamp = timestamp
 	backupFPInfo.SegDirMap = make(map[int]string)
 	backupFPInfo.BaseDataDir = "<SEG_DATA_DIR>"
@@ -51,6 +53,14 @@ func NewFilePathInfo(c *cluster.Cluster, userSpecifiedBackupDir string, timestam
 	return backupFPInfo
 }
 
+/*
+ * Set user specified dir for report.
+ * Currently used for restore only.
+ */
+func (backupFPInfo *FilePathInfo) SetReportDir(userSpecifiedReportDir string) {
+	backupFPInfo.UserSpecifiedReportDir = userSpecifiedReportDir
+}
+
 /*
  * Restoring a future-dated backup is allowed (e.g. the backup was taken in a
  * different time zone that is ahead of the restore time zone), so only check
@@ -73,6 +83,18 @@ func (backupFPInfo *FilePathInfo) GetDirForContent(contentID int) string {
 	return path.Join(backupFPInfo.SegDirMap[contentID], "backups", backupFPInfo.Timestamp[0:8], backupFPInfo.Timestamp)
 }
 
+func (backupFPInfo *FilePathInfo) IsUserSpecifiedReportDir() bool {
+	return backupFPInfo.UserSpecifiedReportDir != ""
+}
+
+func (backupFPInfo *FilePathInfo) GetDirForReport(contentID int) string {
+	if backupFPInfo.IsUserSpecifiedReportDir() {
+		segDir := fmt.Sprintf("%s%d", backupFPInfo.UserSpecifiedSegPrefix, contentID)
+		return path.Join(backupFPInfo.UserSpecifiedReportDir, segDir, "backups", backupFPInfo.Timestamp[0:8], backupFPInfo.Timestamp)
+	}
+	return backupFPInfo.GetDirForContent(contentID);
+}
+
 func (backupFPInfo *FilePathInfo) replaceCopyFormatStringsInPath(templateFilePath string, contentID int) string {
 	filePath := strings.Replace(templateFilePath, "<SEG_DATA_DIR>", backupFPInfo.SegDirMap[contentID], -1)
 	return strings.Replace(filePath, "<SEGID>", strconv.Itoa(contentID), -1)
@@ -148,7 +170,7 @@ func (backupFPInfo *FilePathInfo) GetBackupReportFilePath() string {
 }
 
 func (backupFPInfo *FilePathInfo) GetRestoreFilePath(restoreTimestamp string, filetype string) string {
-	return path.Join(backupFPInfo.GetDirForContent(-1), fmt.Sprintf("gprestore_%s_%s_%s", backupFPInfo.Timestamp, restoreTimestamp, metadataFilenameMap[filetype]))
+	return path.Join(backupFPInfo.GetDirForReport(-1), fmt.Sprintf("gprestore_%s_%s_%s", backupFPInfo.Timestamp, restoreTimestamp, metadataFilenameMap[filetype]))
 }
 
 func (backupFPInfo *FilePathInfo) GetRestoreReportFilePath(restoreTimestamp string) string {

filepath/filepath_test.go

@@ -140,6 +140,20 @@ var _ = Describe("filepath tests", func() {
 			fpInfo := NewFilePathInfo(c, "/foo/bar", "20170101010101", "gpseg")
 			Expect(fpInfo.GetBackupReportFilePath()).To(Equal("/foo/bar/gpseg-1/backups/20170101/20170101010101/gpbackup_20170101010101_report"))
 		})
+		It("returns report file path for restore command", func() {
+			fpInfo := NewFilePathInfo(c, "", "20170101010101", "gpseg")
+			Expect(fpInfo.GetRestoreReportFilePath("20200101010101")).To(Equal("/data/gpseg-1/backups/20170101/20170101010101/gprestore_20170101010101_20200101010101_report"))
+		})
+		It("returns report file path based on user specified path for restore command", func() {
+			fpInfo := NewFilePathInfo(c, "/foo/bar", "20170101010101", "gpseg")
+			Expect(fpInfo.GetRestoreReportFilePath("20200101010101")).To(Equal("/foo/bar/gpseg-1/backups/20170101/20170101010101/gprestore_20170101010101_20200101010101_report"))
+		})
+		It("returns different report file paths based on user specified report path for backup and restore command", func() {
+			fpInfo := NewFilePathInfo(c, "/foo/bar", "20170101010101", "gpseg")
+			fpInfo.SetReportDir("/bar/foo")
+			Expect(fpInfo.GetBackupReportFilePath()).To(Equal("/foo/bar/gpseg-1/backups/20170101/20170101010101/gpbackup_20170101010101_report"))
+			Expect(fpInfo.GetRestoreReportFilePath("20200101010101")).To(Equal("/bar/foo/gpseg-1/backups/20170101/20170101010101/gprestore_20170101010101_20200101010101_report"))
+		})
 	})
 	Describe("GetTableBackupFilePath", func() {
 		It("returns table file path", func() {

options/flag.go

@@ -52,6 +52,7 @@ const (
 	WITHOUT_GLOBALS       = "without-globals"
 	RESIZE_CLUSTER        = "resize-cluster"
 	NO_INHERITS           = "no-inherits"
+	REPORT_DIR            = "report-dir"
 )
 
 func SetBackupFlagDefaults(flagSet *pflag.FlagSet) {
@@ -120,6 +121,7 @@ func SetRestoreFlagDefaults(flagSet *pflag.FlagSet) {
 	flagSet.Bool(LEAF_PARTITION_DATA, false, "For partition tables, create one data file per leaf partition instead of one data file for the whole table")
 	flagSet.Bool(RUN_ANALYZE, false, "Run ANALYZE on restored tables")
 	flagSet.Bool(RESIZE_CLUSTER, false, "Restore a backup taken on a cluster with more or fewer segments than the cluster to which it will be restored")
+	flagSet.String(REPORT_DIR, "", "The absolute path of the directory to which all report files will be written")
 	_ = flagSet.MarkHidden(LEAF_PARTITION_DATA)
 }
 

restore/restore.go

@@ -75,6 +75,11 @@ func DoSetup() {
 	segPrefix, err = filepath.ParseSegPrefix(MustGetFlagString(options.BACKUP_DIR))
 	gplog.FatalOnError(err)
 	globalFPInfo = filepath.NewFilePathInfo(globalCluster, MustGetFlagString(options.BACKUP_DIR), backupTimestamp, segPrefix)
+	if reportDir := MustGetFlagString(options.REPORT_DIR); reportDir != "" {
+		globalFPInfo.SetReportDir(reportDir)
+		info, err := globalCluster.ExecuteLocalCommand(fmt.Sprintf("mkdir -p %s", globalFPInfo.GetDirForReport(-1)))
+		gplog.FatalOnError(err, info)
+	}
 
 	// Get restore metadata from plugin
 	if MustGetFlagString(options.PLUGIN_CONFIG) != "" {