feat: update FindPriorBackups to use backup index

This patch teaches `FindPriorBackups` (now more aptly renamed to
`FindAllIncrementals`) to use the backup index introduced in cockroachdb#150384.
Additional changes will still be required to fully fix the related
tickets, namely updating the logic to no longer need to load ALL
manifests of a chain into memory, only the manifests needed.

Epic: CRDB-47942

Informs: cockroachdb#150327, cockroachdb#150329, cockroachdb#150330, cockroachdb#150331

Author: kev-cao

pkg/backup/backupdest/backup_destination.go

@@ -21,7 +21,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/backup/backuputils"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/cloud/cloudpb"
-	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
@@ -220,7 +219,12 @@ func ResolveDest(
 	}
 	defer incrementalStore.Close()
 
-	priors, err := FindPriorBackups(ctx, incrementalStore, OmitManifest)
+	rootStore, err := makeCloudStorage(ctx, collectionURI, user)
+	if err != nil {
+		return ResolvedDestination{}, err
+	}
+	defer rootStore.Close()
+	priors, err := FindAllIncrementals(ctx, incrementalStore, rootStore, chosenSuffix, OmitManifest)
 	if err != nil {
 		return ResolvedDestination{}, errors.Wrap(err, "adjusting backup destination to append new layer to existing backup")
 	}
@@ -237,41 +241,40 @@ func ResolveDest(
 	}
 	prevBackupURIs = append([]string{plannedBackupDefaultURI}, prevBackupURIs...)
 
-	// Within the chosenSuffix dir, differentiate incremental backups with partName.
-	partName := endTime.GoTime().Format(backupbase.DateBasedIncFolderName)
-	if execCfg.Settings.Version.IsActive(ctx, clusterversion.V25_2) {
-		if startTime.IsEmpty() {
-			baseEncryptionOptions, err := backupencryption.GetEncryptionFromBase(
-				ctx, user, execCfg.DistSQLSrv.ExternalStorageFromURI, prevBackupURIs[0],
-				encryption, kmsEnv,
-			)
-			if err != nil {
-				return ResolvedDestination{}, err
-			}
+	// If startTime is not already set, we will find it via the previous backup
+	// manifest.
+	if startTime.IsEmpty() {
+		baseEncryptionOptions, err := backupencryption.GetEncryptionFromBase(
+			ctx, user, execCfg.DistSQLSrv.ExternalStorageFromURI, prevBackupURIs[0],
+			encryption, kmsEnv,
+		)
+		if err != nil {
+			return ResolvedDestination{}, err
+		}
 
-			// TODO (kev-cao): Once we have completed the backup directory index work, we
-			// can remove the need to read an entire backup manifest just to fetch the
-			// start time. We can instead read the metadata protobuf.
-			mem := execCfg.RootMemoryMonitor.MakeBoundAccount()
-			defer mem.Close(ctx)
-			precedingBackupManifest, size, err := backupinfo.ReadBackupManifestFromURI(
-				ctx, &mem, prevBackupURIs[len(prevBackupURIs)-1], user,
-				execCfg.DistSQLSrv.ExternalStorageFromURI, baseEncryptionOptions, kmsEnv,
-			)
-			if err != nil {
-				return ResolvedDestination{}, err
-			}
-			if err := mem.Grow(ctx, size); err != nil {
-				return ResolvedDestination{}, err
-			}
-			defer mem.Shrink(ctx, size)
-			startTime = precedingBackupManifest.EndTime
-			if startTime.IsEmpty() {
-				return ResolvedDestination{}, errors.Errorf("empty end time in prior backup manifest")
-			}
+		// TODO (kev-cao): Once we have completed the backup directory index work, we
+		// can remove the need to read an entire backup manifest just to fetch the
+		// start time. We can instead read the metadata protobuf.
+		mem := execCfg.RootMemoryMonitor.MakeBoundAccount()
+		defer mem.Close(ctx)
+		precedingBackupManifest, size, err := backupinfo.ReadBackupManifestFromURI(
+			ctx, &mem, prevBackupURIs[len(prevBackupURIs)-1], user,
+			execCfg.DistSQLSrv.ExternalStorageFromURI, baseEncryptionOptions, kmsEnv,
+		)
+		if err != nil {
+			return ResolvedDestination{}, err
+		}
+		if err := mem.Grow(ctx, size); err != nil {
+			return ResolvedDestination{}, err
+		}
+		defer mem.Shrink(ctx, size)
+		startTime = precedingBackupManifest.EndTime
+		if startTime.IsEmpty() {
+			return ResolvedDestination{}, errors.Errorf("empty end time in prior backup manifest")
 		}
-		partName = partName + "-" + startTime.GoTime().Format(backupbase.DateBasedIncFolderNameSuffix)
 	}
+
+	partName := ConstructDateBasedIncrementalFolderName(startTime.GoTime(), endTime.GoTime())
 	defaultIncrementalsURI, urisByLocalityKV, err := GetURIsByLocalityKV(fullyResolvedIncrementalsLocation, partName)
 	if err != nil {
 		return ResolvedDestination{}, err
@@ -540,9 +543,11 @@ func ListFullBackupsInCollection(
 func ResolveBackupManifests(
 	ctx context.Context,
 	mem *mon.BoundAccount,
+	defaultCollectionURI string,
 	baseStores []cloud.ExternalStorage,
 	incStores []cloud.ExternalStorage,
 	mkStore cloud.ExternalStorageFromURIFactory,
+	resolvedSubdir string,
 	fullyResolvedBaseDirectory []string,
 	fullyResolvedIncrementalsDirectory []string,
 	endTime hlc.Timestamp,
@@ -577,7 +582,14 @@ func ResolveBackupManifests(
 
 	var incrementalBackups []string
 	if len(incStores) > 0 {
-		incrementalBackups, err = FindPriorBackups(ctx, incStores[0], includeManifest)
+		rootStore, err := mkStore(ctx, defaultCollectionURI, user)
+		if err != nil {
+			return nil, nil, nil, 0, err
+		}
+		defer rootStore.Close()
+		incrementalBackups, err = FindAllIncrementals(
+			ctx, incStores[0], rootStore, resolvedSubdir, includeManifest,
+		)
 		if err != nil {
 			return nil, nil, nil, 0, err
 		}

pkg/backup/backupdest/backup_destination_test.go

@@ -111,6 +111,7 @@ func TestBackupRestoreResolveDestination(t *testing.T) {
 			// - BACKUP INTO full1 IN collection, incremental_location = inc_storage_path @ 9
 			// - BACKUP INTO full1 IN collection, incremental_location = inc_storage_path @ 9:30
 			// - BACKUP INTO LATEST IN collection, incremental_location = inc_storage_path @ 10
+			// - BACKUP INTO collection (full 3) @ 10:30
 			t.Run("collection", func(t *testing.T) {
 				collectionLoc := fmt.Sprintf("nodelocal://1/%s?AUTH=implicit", t.Name())
 				// Note that this default is NOT arbitrary, but rather hard-coded as
@@ -134,9 +135,6 @@ func TestBackupRestoreResolveDestination(t *testing.T) {
 				inc6Time := inc5Time.Add(time.Minute * 30)
 				inc7Time := inc6Time.Add(time.Minute * 30)
 				full3Time := inc7Time.Add(time.Minute * 30)
-				inc8Time := full3Time.Add(time.Minute * 30)
-				inc9Time := inc8Time.Add(time.Minute * 30)
-				full4Time := inc9Time.Add(time.Minute * 30)
 
 				// firstBackupChain is maintained throughout the tests as the history of
 				// backups that were taken based on the initial full backup.
@@ -155,6 +153,7 @@ func TestBackupRestoreResolveDestination(t *testing.T) {
 				testCollectionBackup := func(t *testing.T, backupTime time.Time,
 					expectedDefault, expectedSuffix, expectedIncDir string, expectedPrevBackups []string,
 					appendToLatest bool, subdir string, incrementalTo []string) {
+					t.Helper()
 
 					endTime := hlc.Timestamp{WallTime: backupTime.UnixNano()}
 
@@ -335,14 +334,11 @@ func TestBackupRestoreResolveDestination(t *testing.T) {
 						true /* intoLatest */, expectedSubdir, customIncrementalTo)
 					writeManifest(t, expectedDefault, inc7Time)
 				}
-
 				// A new full backup: BACKUP INTO collection
-				var backup3Location string
 				{
 					expectedSuffix := "/2020/12/25-103000.00"
 					expectedIncDir := ""
 					expectedDefault := fmt.Sprintf("nodelocal://1/%s%s?AUTH=implicit", t.Name(), expectedSuffix)
-					backup3Location = expectedDefault
 
 					testCollectionBackup(t, full3Time,
 						expectedDefault, expectedSuffix, expectedIncDir, []string(nil),
@@ -351,47 +347,6 @@ func TestBackupRestoreResolveDestination(t *testing.T) {
 					// We also wrote a new full backup, so let's update the latest.
 					writeLatest(t, collectionLoc, expectedSuffix)
 				}
-
-				// A remote incremental into the third full backup: BACKUP INTO LATEST
-				// IN collection, BUT with a trick. Write a (fake) incremental backup
-				// to the old directory, to be sure that subsequent incremental backups
-				// go there as well (and not the newer incrementals/ subdir.)
-				{
-					expectedSuffix := "/2020/12/25-103000.00"
-					expectedIncDir := "/20201225/110000.00-20201225-103000.00"
-
-					// Writes the (fake) incremental backup.
-					oldStyleDefault := fmt.Sprintf("nodelocal://1/%s%s%s?AUTH=implicit",
-						t.Name(),
-						expectedSuffix, expectedIncDir)
-					writeManifest(t, oldStyleDefault, inc8Time)
-
-					expectedSuffix = "/2020/12/25-103000.00"
-					expectedIncDir = "/20201225/113000.00-20201225-110000.00"
-					expectedSubdir := expectedSuffix
-					expectedDefault := fmt.Sprintf("nodelocal://1/%s%s%s?AUTH=implicit",
-						t.Name(),
-						expectedSuffix, expectedIncDir)
-
-					testCollectionBackup(t, inc9Time,
-						expectedDefault, expectedSuffix, expectedIncDir, []string{backup3Location, oldStyleDefault},
-						true /* intoLatest */, expectedSubdir, noIncrementalStorage)
-					writeManifest(t, expectedDefault, inc9Time)
-				}
-
-				// A new full backup: BACKUP INTO collection
-				{
-					expectedSuffix := "/2020/12/25-120000.00"
-					expectedIncDir := ""
-					expectedDefault := fmt.Sprintf("nodelocal://1/%s%s?AUTH=implicit", t.Name(), expectedSuffix)
-
-					testCollectionBackup(t, full4Time,
-						expectedDefault, expectedSuffix, expectedIncDir, []string(nil),
-						false /* intoLatest */, noExplicitSubDir, noIncrementalStorage)
-					writeManifest(t, expectedDefault, full4Time)
-					// We also wrote a new full backup, so let's update the latest.
-					writeLatest(t, collectionLoc, expectedSuffix)
-				}
 			})
 		})
 	}

pkg/backup/backupdest/backup_index.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"path"
 	"strings"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/backup/backupbase"
 	"github.com/cockroachdb/cockroach/pkg/backup/backuppb"
@@ -120,10 +121,9 @@ func WriteBackupIndexMetadata(
 // can remove these checks in v26.2+.
 func IndexExists(ctx context.Context, store cloud.ExternalStorage, subdir string) (bool, error) {
 	var indexExists bool
-	indexSubdir := path.Join(backupbase.BackupIndexDirectoryPath, flattenSubdirForIndex(subdir))
 	if err := store.List(
 		ctx,
-		indexSubdir,
+		indexSubdir(subdir),
 		"/",
 		func(file string) error {
 			indexExists = true
@@ -135,6 +135,83 @@ func IndexExists(ctx context.Context, store cloud.ExternalStorage, subdir string
 	return indexExists, nil
 }
 
+// ListIndexes lists all the index files for a backup chain rooted by the full
+// backup indicated by the subdir. The store should be rooted at the default
+// collection URI (the one that contains the `index/` directory). It returns
+// the basenames of the listed index files. It assumes that the subdir is
+// resolved and not `LATEST`.
+//
+// The indexes are returned in reverse chronological order, i.e. the most recent
+// index is first in the list.
+func ListIndexes(
+	ctx context.Context, store cloud.ExternalStorage, subdir string,
+) ([]string, error) {
+	var indexBasenames []string
+	if err := store.List(
+		ctx,
+		indexSubdir(subdir),
+		"",
+		func(file string) error {
+			indexBasenames = append(indexBasenames, path.Base(file))
+			return nil
+		},
+	); err != nil {
+		return nil, errors.Wrapf(err, "listing indexes in %s", subdir)
+	}
+	return indexBasenames, nil
+}
+
+// ParseIndexFilename parses the start and end timestamps from the index
+// filename.
+//
+// Note: The timestamps are only millisecond-precise and so do not represent the
+// exact nano-specific times in the corresponding backup manifest.
+func parseIndexFilename(basename string) (start time.Time, end time.Time, err error) {
+	invalidFmtErr := errors.Newf("invalid index filename format: %s", basename)
+
+	if !strings.HasSuffix(basename, "_metadata.pb") {
+		return time.Time{}, time.Time{}, invalidFmtErr
+	}
+	parts := strings.Split(basename, "_")
+	if len(parts) != 4 {
+		return time.Time{}, time.Time{}, invalidFmtErr
+	}
+
+	if parts[1] != "0" {
+		start, err = time.Parse(backupbase.BackupIndexFilenameTimestampFormat, parts[1])
+		if err != nil {
+			return time.Time{}, time.Time{}, errors.Join(invalidFmtErr, err)
+		}
+	}
+	end, err = time.Parse(backupbase.BackupIndexFilenameTimestampFormat, parts[2])
+	if err != nil {
+		return time.Time{}, time.Time{}, errors.Join(invalidFmtErr, err)
+	}
+
+	return start, end, nil
+}
+
+// ParseBackupFilePathFromIndexFileName parses the path to a backup given the
+// basename of its index file and its subdirectory. For full backups, the
+// returned path is relative to the root of the backup. For incremental
+// backups, the path is relative to the incremental storage location. It expects
+// that subdir is resolved and not `LATEST`.
+//
+// Note: While the path is stored in the index file, we can take a shortcut here
+// and derive it from the filename solely because backup paths are
+// millisecond-precise and so are the timestamps encoded in the filename.
+func parseBackupFilePathFromIndexFileName(subdir, basename string) (string, error) {
+	start, end, err := parseIndexFilename(basename)
+	if err != nil {
+		return "", err
+	}
+	if start.IsZero() {
+		return subdir, nil
+	}
+
+	return ConstructDateBasedIncrementalFolderName(start, end), nil
+}
+
 // shouldWriteIndex determines if a backup index file should be written for a
 // given backup. The rule is:
 //  1. An index should only be written on a v25.4+ cluster.
@@ -172,8 +249,7 @@ func getBackupIndexFilePath(subdir string, startTime, endTime hlc.Timestamp) (st
 		return "", errors.AssertionFailedf("expected subdir to be resolved and not be 'LATEST'")
 	}
 	return backuputils.JoinURLPath(
-		backupbase.BackupIndexDirectoryPath,
-		flattenSubdirForIndex(subdir),
+		indexSubdir(subdir),
 		getBackupIndexFileName(startTime, endTime),
 	), nil
 }
@@ -194,13 +270,15 @@ func getBackupIndexFileName(startTime, endTime hlc.Timestamp) string {
 	)
 }
 
-// flattenSubdirForIndex flattens a full backup subdirectory to be used in the
-// index. It assumes subdir is not `LATEST` and has been resolved.
-// We flatten the subdir so that when listing from the index, we can list with
-// the `index/` prefix and delimit on `/`.
-func flattenSubdirForIndex(subdir string) string {
-	return strings.ReplaceAll(
+// indexSubdir returns the path to the index directory for a given full backup
+// subdir relative to the root of the default collection URI. It assumes that
+// subdir is not `LATEST` and has been resolved.
+func indexSubdir(subdir string) string {
+	// We flatten the subdir so that when listing from the index, we can list with
+	// the `index/` prefix and delimit on `/`.
+	flattenedSubdir := strings.ReplaceAll(
 		strings.TrimPrefix(subdir, "/"),
 		"/", "-",
 	)
+	return path.Join(backupbase.BackupIndexDirectoryPath, flattenedSubdir)
 }