backup: conditionally write backup index for mixed-version

A backup index file is only usable under the following conditions:

1. An index file is written for the full backup of the chain it belongs
to.
2. Every incremental in the chain wrote an index.

For condition 1, we can ensure this by checking that for incremental
backups, the index already exists. In the case of full backups, we can
jsut write the index.

However, for 2, we need to ensure that the entire cluster is on v25.4+.
Otherwise, it is possible for a full backup to be coordinated by a node
that was on 25.4+ and therefore write an index, but for a following
incremental to be written by a pre-25.4 node. So we only write indexes
if the entire cluster is v25.4+.

Epic: CRDB-47942

Release note: None

Author: kev-cao

pkg/backup/backup_job.go

@@ -441,6 +441,7 @@ func backup(
 
 	if err := backupdest.WriteBackupIndexMetadata(
 		ctx,
+		execCtx.ExecCfg(),
 		execCtx.User(),
 		execCtx.ExecCfg().DistSQLSrv.ExternalStorageFromURI,
 		details,

pkg/backup/backupdest/BUILD.bazel

@@ -60,11 +60,13 @@ go_test(
         "//pkg/cloud",
         "//pkg/cloud/cloudpb",
         "//pkg/cloud/impl:cloudimpl",
+        "//pkg/clusterversion",
         "//pkg/jobs/jobspb",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
         "//pkg/security/username",
         "//pkg/server",
+        "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",

pkg/backup/backupdest/backup_index.go

@@ -9,14 +9,17 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"path"
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/backup/backupbase"
 	"github.com/cockroachdb/cockroach/pkg/backup/backuppb"
 	"github.com/cockroachdb/cockroach/pkg/backup/backuputils"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
@@ -31,11 +34,25 @@ import (
 // information.
 func WriteBackupIndexMetadata(
 	ctx context.Context,
+	execCfg *sql.ExecutorConfig,
 	user username.SQLUsername,
 	makeExternalStorageFromURI cloud.ExternalStorageFromURIFactory,
 	details jobspb.BackupDetails,
 ) error {
-	ctx, sp := tracing.ChildSpan(ctx, "backupdest.WriteBackupIndexMetadata")
+	indexStore, err := makeExternalStorageFromURI(
+		ctx, details.CollectionURI, user,
+	)
+	if err != nil {
+		return errors.Wrapf(err, "creating external storage")
+	}
+
+	if shouldWrite, err := shouldWriteIndex(
+		ctx, execCfg, indexStore, details,
+	); !shouldWrite {
+		return err
+	}
+
+	ctx, sp := tracing.ChildSpan(ctx, "backupinfo.WriteBackupIndexMetadata")
 	defer sp.Finish()
 
 	if details.EndTime.IsEmpty() {
@@ -74,13 +91,6 @@ func WriteBackupIndexMetadata(
 		return errors.Wrapf(err, "marshal backup index metadata")
 	}
 
-	indexStore, err := makeExternalStorageFromURI(
-		ctx, details.CollectionURI, user,
-	)
-	if err != nil {
-		return errors.Wrapf(err, "creating external storage")
-	}
-
 	indexFilePath, err := getBackupIndexFilePath(
 		details.Destination.Subdir,
 		details.StartTime,
@@ -95,22 +105,75 @@ func WriteBackupIndexMetadata(
 	)
 }
 
+// IndexExists checks if for a given full backup subdirectory there exists a
+// corresponding index in the backup collection. This is used to determine when
+// we should use the index or the legacy path.
+//
+// This works under the assumption that we only ever write an index iff:
+//  1. For an incremental backup, an index exists for its full backup.
+//  2. The backup was taken on a v25.4+ cluster.
+//
+// The store should be rooted at the default collection URI (the one that
+// contains the `index/` directory).
+//
+// Note: v25.4+ backups will always contain an index file. In other words, we
+// 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,
+		"/",
+		func(file string) error {
+			indexExists = true
+			return errors.New("found index")
+		},
+	); err != nil && !indexExists {
+		return false, errors.Wrapf(err, "checking index exists in %s", subdir)
+	}
+	return indexExists, 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.
+//  2. An incremental backup only writes an index if its parent full has written
+//     an index file.
+//
+// This ensures that if a backup chain exists in the index directory, then every
+// backup in that chain has an index file, ensuring that the index is usable.
+func shouldWriteIndex(
+	ctx context.Context,
+	execCfg *sql.ExecutorConfig,
+	store cloud.ExternalStorage,
+	details jobspb.BackupDetails,
+) (bool, error) {
+	// This version check can be removed in v26.1 when we no longer need to worry
+	// about a mixed-version cluster where we have both v25.4+ nodes and pre-v25.4
+	// nodes.
+	if !execCfg.Settings.Version.IsActive(ctx, clusterversion.V25_4) {
+		return false, nil
+	}
+
+	// Full backups can write an index as long as the cluster is on v25.4+.
+	if details.StartTime.IsEmpty() {
+		return true, nil
+	}
+
+	return IndexExists(ctx, store, details.Destination.Subdir)
+}
+
 // getBackupIndexFilePath returns the path to the backup index file representing
 // a backup that starts and ends at the given timestamps, including
 // the filename and extension. The path is relative to the collection URI.
 func getBackupIndexFilePath(subdir string, startTime, endTime hlc.Timestamp) (string, error) {
 	if strings.EqualFold(subdir, backupbase.LatestFileName) {
 		return "", errors.AssertionFailedf("expected subdir to be resolved and not be 'LATEST'")
 	}
-	// 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 backuputils.JoinURLPath(
 		backupbase.BackupIndexDirectoryPath,
-		flattenedSubdir,
+		flattenSubdirForIndex(subdir),
 		getBackupIndexFileName(startTime, endTime),
 	), nil
 }
@@ -130,3 +193,14 @@ func getBackupIndexFileName(startTime, endTime hlc.Timestamp) string {
 		descEndTs, formattedStartTime, formattedEndTime,
 	)
 }
+
+// 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(
+		strings.TrimPrefix(subdir, "/"),
+		"/", "-",
+	)
+}