Merge #153065 #153380 #153470

153065: restore: scale number of OR link workers to number of nodes r=msbutler a=kev-cao

This patch teaches online restore to scale the number of link workers to the number of nodes in the cluster instead of statically setting it to 32. In the new behavior, the default value for `backup.restore.online_worker_count` is 0, which sets the number of link workers to two times the number of nodes.

Epic: CRDB-48786

Fixes: #146584

153380: sql: fix node panic on SIGINT during CHECK EXTERNAL CONNECTION r=jeffswenson a=kev-cao

`CHECK EXTERNAL CONNECTION` defers a close of the `rows` channel as part of its execution. It also closes that same channel in its `Close` method. Under normal execution, if `CHECK EXTERNAL CONNECTION` is allowed to complete, that `rows` channel is set to `nil`, so `Close` skips the attempt to close to the channel.

However, if a `SIGINT` is sent during the execution to cancel the query, `rows` is never set to `nil` and `Close` will attempt to close a closed channel, causing a node panic.

Epic: None

Release note (bug fix): Fixed a bug introduced in v25.1.0 that would cause a node panic if a `SIGINT` signal was sent during the execution of a `CHECK EXTERNAL CONNECTION` command.

153470: backupdest: attempt to close all stores in MakeBackupDestinationStores cleanupFn r=kev-cao a=msbutler



Epic: none

Release note: none

Co-authored-by: Kevin Cao <[email protected]>
Co-authored-by: Michael Butler <[email protected]>

Author: 3 people

pkg/backup/backupdest/backup_destination.go

@@ -557,8 +557,6 @@ func ResolveBackupManifests(
 	mem *mon.BoundAccount,
 	defaultCollectionURI string,
 	collectionURIs []string,
-	baseStores []cloud.ExternalStorage,
-	incStores []cloud.ExternalStorage,
 	mkStore cloud.ExternalStorageFromURIFactory,
 	resolvedSubdir string,
 	fullyResolvedBaseDirectory []string,
@@ -590,7 +588,7 @@ func ResolveBackupManifests(
 
 	if !ReadBackupIndexEnabled.Get(&execCfg.Settings.SV) || !exists || isCustomIncLocation {
 		return legacyResolveBackupManifests(
-			ctx, execCfg, mem, defaultCollectionURI, baseStores, incStores, mkStore,
+			ctx, execCfg, mem, defaultCollectionURI, mkStore,
 			resolvedSubdir, fullyResolvedBaseDirectory, fullyResolvedIncrementalsDirectory,
 			endTime, encryption, kmsEnv, user, includeSkipped, includeCompacted,
 		)
@@ -609,8 +607,6 @@ func legacyResolveBackupManifests(
 	execCfg *sql.ExecutorConfig,
 	mem *mon.BoundAccount,
 	defaultCollectionURI string,
-	baseStores []cloud.ExternalStorage,
-	incStores []cloud.ExternalStorage,
 	mkStore cloud.ExternalStorageFromURIFactory,
 	resolvedSubdir string,
 	fullyResolvedBaseDirectory []string,
@@ -637,6 +633,27 @@ func legacyResolveBackupManifests(
 			mem.Shrink(ctx, ownedMemSize)
 		}
 	}()
+
+	baseStores, baseCleanupFn, err := MakeBackupDestinationStores(ctx, user, mkStore, fullyResolvedBaseDirectory)
+	if err != nil {
+		return nil, nil, nil, 0, err
+	}
+	defer func() {
+		if err := baseCleanupFn(); err != nil {
+			log.Dev.Warningf(ctx, "failed to close base store: %+v", err)
+		}
+	}()
+
+	incStores, incCleanupFn, err := MakeBackupDestinationStores(ctx, user, mkStore, fullyResolvedIncrementalsDirectory)
+	if err != nil {
+		return nil, nil, nil, 0, err
+	}
+	defer func() {
+		if err := incCleanupFn(); err != nil {
+			log.Dev.Warningf(ctx, "failed to close inc store: %+v", err)
+		}
+	}()
+
 	baseManifest, memSize, err := backupinfo.ReadBackupManifestFromStore(ctx, mem, baseStores[0], fullyResolvedBaseDirectory[0],
 		encryption, kmsEnv)
 	if err != nil {
@@ -646,11 +663,6 @@ func legacyResolveBackupManifests(
 
 	var incrementalBackups []string
 	if len(incStores) > 0 {
-		rootStore, err := mkStore(ctx, defaultCollectionURI, user)
-		if err != nil {
-			return nil, nil, nil, 0, err
-		}
-		defer rootStore.Close()
 		incrementalBackups, err = LegacyFindPriorBackups(ctx, incStores[0], includeManifest)
 		if err != nil {
 			return nil, nil, nil, 0, err
@@ -771,23 +783,15 @@ func indexedResolveBackupManifests(
 	ctx, sp := tracing.ChildSpan(ctx, "backupdest.ResolveBackupManifestsWithIndexes")
 	defer sp.Finish()
 
-	rootStores := make([]cloud.ExternalStorage, 0, len(collectionURIs))
+	rootStores, rootCleanupFn, err := MakeBackupDestinationStores(ctx, user, mkStore, collectionURIs)
+	if err != nil {
+		return nil, nil, nil, 0, err
+	}
 	defer func() {
-		for _, store := range rootStores {
-			if store != nil {
-				if err := store.Close(); err != nil {
-					log.Dev.Errorf(ctx, "error closing store: %v", err)
-				}
-			}
+		if err := rootCleanupFn(); err != nil {
+			log.Dev.Warningf(ctx, "failed to close collection store: %s", err)
 		}
 	}()
-	for _, uri := range collectionURIs {
-		store, err := mkStore(ctx, uri, user)
-		if err != nil {
-			return nil, nil, nil, 0, err
-		}
-		rootStores = append(rootStores, store)
-	}
 
 	indexes, err := GetBackupTreeIndexMetadata(
 		ctx, rootStores[0], resolvedSubdir,

pkg/backup/backupdest/backup_destination_test.go

@@ -396,31 +396,13 @@ func TestResolveBackupManifests(t *testing.T) {
 	})
 	require.NoError(t, err)
 
-	baseStores, err := util.MapE(fullyResolvedBaseDirs, func(uri string) (cloud.ExternalStorage, error) {
-		return execCfg.DistSQLSrv.ExternalStorageFromURI(ctx, uri, username.RootUserName())
-	})
-	require.NoError(t, err)
-	for i := range baseStores {
-		defer baseStores[i].Close()
-	}
-
-	incStores, err := util.MapE(fullyResolvedIncDirs, func(uri string) (cloud.ExternalStorage, error) {
-		return execCfg.DistSQLSrv.ExternalStorageFromURI(ctx, uri, username.RootUserName())
-	})
-	require.NoError(t, err)
-	for i := range incStores {
-		defer incStores[i].Close()
-	}
-
 	t.Run("resolve backup manifests with latest AOST", func(t *testing.T) {
 		uris, manifests, locality, memSize, err := ResolveBackupManifests(
 			ctx,
 			&execCfg,
 			&mem,
 			collections[0],
 			collections,
-			baseStores,
-			incStores,
 			execCfg.DistSQLSrv.ExternalStorageFromURI,
 			fullSubdir,
 			fullyResolvedBaseDirs,
@@ -448,8 +430,6 @@ func TestResolveBackupManifests(t *testing.T) {
 			&mem,
 			collections[0],
 			collections,
-			baseStores,
-			incStores,
 			execCfg.DistSQLSrv.ExternalStorageFromURI,
 			fullSubdir,
 			fullyResolvedBaseDirs,

pkg/backup/backupdest/incrementals.go

@@ -216,23 +216,24 @@ func MakeBackupDestinationStores(
 	mkStore cloud.ExternalStorageFromURIFactory,
 	destinationDirs []string,
 ) ([]cloud.ExternalStorage, func() error, error) {
-	incStores := make([]cloud.ExternalStorage, len(destinationDirs))
+	stores := make([]cloud.ExternalStorage, len(destinationDirs))
 	for i := range destinationDirs {
 		store, err := mkStore(ctx, destinationDirs[i], user)
 		if err != nil {
 			return nil, nil, errors.Wrapf(err, "failed to open backup storage location")
 		}
-		incStores[i] = store
+		stores[i] = store
 	}
 
-	return incStores, func() error {
-		// Close all the incremental stores in the returned cleanup function.
-		for _, store := range incStores {
+	return stores, func() error {
+		// Close all the stores in the returned cleanup function.
+		var combinedErr error
+		for _, store := range stores {
 			if err := store.Close(); err != nil {
-				return err
+				combinedErr = errors.CombineErrors(combinedErr, err)
 			}
 		}
-		return nil
+		return combinedErr
 	}, nil
 }
 

pkg/backup/compaction_job.go

@@ -745,17 +745,6 @@ func getBackupChain(
 			log.Dev.Warningf(ctx, "failed to cleanup base backup stores: %+v", err)
 		}
 	}()
-	incStores, incCleanup, err := backupdest.MakeBackupDestinationStores(
-		ctx, user, mkStore, resolvedIncDirs,
-	)
-	if err != nil {
-		return nil, nil, nil, nil, err
-	}
-	defer func() {
-		if err := incCleanup(); err != nil {
-			log.Dev.Warningf(ctx, "failed to cleanup incremental backup stores: %+v", err)
-		}
-	}()
 	baseEncryptionInfo := encryptionOpts
 	if encryptionOpts != nil && !encryptionOpts.HasKey() {
 		baseEncryptionInfo, err = backupencryption.GetEncryptionFromBaseStore(
@@ -770,7 +759,7 @@ func getBackupChain(
 	defer mem.Close(ctx)
 
 	_, manifests, localityInfo, memReserved, err := backupdest.ResolveBackupManifests(
-		ctx, execCfg, &mem, defaultCollectionURI, dest.To, baseStores, incStores, mkStore, resolvedSubdir,
+		ctx, execCfg, &mem, defaultCollectionURI, dest.To, mkStore, resolvedSubdir,
 		resolvedBaseDirs, resolvedIncDirs, endTime, baseEncryptionInfo, kmsEnv,
 		user, false /*includeSkipped */, true /*includeCompacted */, len(dest.IncrementalStorage) > 0,
 	)

pkg/backup/restore_online.go

@@ -46,12 +46,14 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
-var onlineRestoreLinkWorkers = settings.RegisterByteSizeSetting(
+const defaultLinkWorkersPerNode = 2
+
+var onlineRestoreLinkWorkers = settings.RegisterIntSetting(
 	settings.ApplicationLevel,
 	"backup.restore.online_worker_count",
-	"workers to use for online restore link phase",
-	32,
-	settings.PositiveInt,
+	"total workers to use for online restore link phase (defaults to 2x number of nodes if set to 0)",
+	0,
+	settings.NonNegativeInt,
 )
 
 var onlineRestoreLayerLimit = settings.RegisterIntSetting(
@@ -83,7 +85,10 @@ func splitAndScatter(
 
 	log.Dev.Infof(ctx, "splitting and scattering spans")
 
-	workers := int(onlineRestoreLinkWorkers.Get(&execCtx.ExecCfg().Settings.SV))
+	workers, err := getNumOnlineRestoreLinkWorkers(ctx, execCtx)
+	if err != nil {
+		return err
+	}
 	toScatter := make(chan execinfrapb.RestoreSpanEntry, 1)
 	toSplit := make(chan execinfrapb.RestoreSpanEntry, workers)
 
@@ -270,7 +275,10 @@ func linkExternalFiles(
 
 	log.Dev.Infof(ctx, "ingesting remote files")
 
-	workers := int(onlineRestoreLinkWorkers.Get(&execCtx.ExecCfg().Settings.SV))
+	workers, err := getNumOnlineRestoreLinkWorkers(ctx, execCtx)
+	if err != nil {
+		return 0, 0, err
+	}
 
 	grp := ctxgroup.WithContext(ctx)
 	ch := make(chan execinfrapb.RestoreSpanEntry, workers)
@@ -1039,3 +1047,20 @@ func (r *restoreResumer) maybeCleanupFailedOnlineRestore(
 
 	return unstickRestoreSpans(ctx, p.ExecCfg(), details.DownloadSpans)
 }
+
+// getNumOnlineRestoreLinkWorkers returns the total number of workers to use for
+// the link phase of an online restore.
+func getNumOnlineRestoreLinkWorkers(ctx context.Context, execCtx sql.JobExecContext) (int, error) {
+	if workers := onlineRestoreLinkWorkers.Get(&execCtx.ExecCfg().Settings.SV); workers > 0 {
+		return int(workers), nil
+	}
+	// All nodes are used in a restore
+	_, sqlInstanceIDs, err := execCtx.ExecCfg().DistSQLPlanner.SetupAllNodesPlanning(
+		ctx, execCtx.ExtendedEvalContext(), execCtx.ExecCfg(),
+	)
+	if err != nil {
+		return 0, err
+	}
+	numNodes := min(len(sqlInstanceIDs), 1)
+	return defaultLinkWorkersPerNode * numNodes, nil
+}

pkg/backup/restore_planning.go

@@ -1738,18 +1738,7 @@ func doRestorePlan(
 	}
 	defer func() {
 		if err := cleanupFn(); err != nil {
-			log.Dev.Warningf(ctx, "failed to close incremental store: %+v", err)
-		}
-	}()
-
-	incStores, cleanupFn, err := backupdest.MakeBackupDestinationStores(ctx, p.User(), mkStore,
-		fullyResolvedIncrementalsDirectory)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err := cleanupFn(); err != nil {
-			log.Dev.Warningf(ctx, "failed to close incremental store: %+v", err)
+			log.Dev.Warningf(ctx, "failed to close base store: %+v", err)
 		}
 	}()
 
@@ -1805,7 +1794,7 @@ func doRestorePlan(
 	// directories, return the URIs and manifests of all backup layers in all
 	// localities. Incrementals will be searched for automatically.
 	defaultURIs, mainBackupManifests, localityInfo, memReserved, err := backupdest.ResolveBackupManifests(
-		ctx, p.ExecCfg(), &mem, defaultCollectionURI, from, baseStores, incStores, mkStore,
+		ctx, p.ExecCfg(), &mem, defaultCollectionURI, from, mkStore,
 		fullyResolvedSubdir, fullyResolvedBaseDirectory, fullyResolvedIncrementalsDirectory, endTime,
 		encryption, &kmsEnv, p.User(), false, includeCompacted, len(incFrom) > 0,
 	)

pkg/backup/show.go

@@ -346,20 +346,10 @@ you must pass the 'encryption_info_dir' parameter that points to the directory o
 		info.enc = encryption
 
 		mkStore := p.ExecCfg().DistSQLSrv.ExternalStorageFromURI
-		incStores, cleanupFn, err := backupdest.MakeBackupDestinationStores(ctx, p.User(), mkStore,
-			fullyResolvedIncrementalsDirectory)
-		if err != nil {
-			return err
-		}
-		defer func() {
-			if err := cleanupFn(); err != nil {
-				log.Dev.Warningf(ctx, "failed to close incremental store: %+v", err)
-			}
-		}()
 
 		info.defaultURIs, info.manifests, info.localityInfo, memReserved,
 			err = backupdest.ResolveBackupManifests(
-			ctx, p.ExecCfg(), &mem, defaultCollectionURI, dest, baseStores, incStores, mkStore, subdir,
+			ctx, p.ExecCfg(), &mem, defaultCollectionURI, dest, mkStore, subdir,
 			fullyResolvedDest, fullyResolvedIncrementalsDirectory, hlc.Timestamp{},
 			encryption, &kmsEnv, p.User(), true /* includeSkipped */, true, /* includeCompacted */
 			len(explicitIncPaths) > 0,

pkg/sql/check_external_connection.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
@@ -26,6 +27,7 @@ func (p *planner) CheckExternalConnection(
 
 type checkExternalConnectionNode struct {
 	zeroInputPlanNode
+	execGrp ctxgroup.Group
 	node    *tree.CheckExternalConnection
 	loc     string
 	params  CloudCheckParams
@@ -101,7 +103,9 @@ func (n *checkExternalConnectionNode) startExec(params runParams) error {
 		return nil
 	})
 
-	go func() {
+	grp := ctxgroup.WithContext(ctx)
+	n.execGrp = grp
+	grp.GoCtx(func(ctx context.Context) error {
 		recv := MakeDistSQLReceiver(
 			ctx,
 			rowWriter,
@@ -117,7 +121,9 @@ func (n *checkExternalConnectionNode) startExec(params runParams) error {
 		// Copy the eval.Context, as dsp.Run() might change it.
 		evalCtxCopy := params.extendedEvalCtx.Context.Copy()
 		dsp.Run(ctx, planCtx, nil, plan, recv, evalCtxCopy, nil /* finishedSetupFn */)
-	}()
+		return nil
+	})
+
 	return nil
 }
 
@@ -130,7 +136,6 @@ func (n *checkExternalConnectionNode) Next(params runParams) (bool, error) {
 		return false, params.ctx.Err()
 	case row, more := <-n.rows:
 		if !more {
-			n.rows = nil
 			return false, nil
 		}
 		n.row = row
@@ -143,10 +148,8 @@ func (n *checkExternalConnectionNode) Values() tree.Datums {
 }
 
 func (n *checkExternalConnectionNode) Close(_ context.Context) {
-	if n.rows != nil {
-		close(n.rows)
-		n.rows = nil
-	}
+	_ = n.execGrp.Wait()
+	n.rows = nil
 }
 
 func (n *checkExternalConnectionNode) parseParams(params runParams) error {