backup: integrate hot loop AC into backup compactions

Fixes: #143903

Release note: None

Author: kev-cao

pkg/backup/backup_processor.go

@@ -444,23 +444,9 @@ func runBackupProcessor(
 
 		readTime := spec.BackupEndTime.GoTime()
 
-		// Passing a nil pacer is effectively a noop if CPU control is disabled.
-		var pacer *admission.Pacer = nil
-		if fileSSTSinkElasticCPUControlEnabled.Get(&clusterSettings.SV) {
-			tenantID, ok := roachpb.ClientTenantFromContext(ctx)
-			if !ok {
-				tenantID = roachpb.SystemTenantID
-			}
-			pacer = flowCtx.Cfg.AdmissionPacerFactory.NewPacer(
-				100*time.Millisecond,
-				admission.WorkInfo{
-					TenantID:        tenantID,
-					Priority:        admissionpb.BulkNormalPri,
-					CreateTime:      timeutil.Now().UnixNano(),
-					BypassAdmission: false,
-				},
-			)
-		}
+		pacer := newBackupPacer(
+			ctx, flowCtx.Cfg.AdmissionPacerFactory, clusterSettings,
+		)
 		// It is safe to close a nil pacer.
 		defer pacer.Close()
 
@@ -774,6 +760,30 @@ func logClose(ctx context.Context, c io.Closer, desc string) {
 	}
 }
 
+// newBackupPacer creates a new AC pacer for backup. It may return nil if CPU
+// control is disabled, which is effectively a noop.
+func newBackupPacer(
+	ctx context.Context, factory admission.PacerFactory, settings *cluster.Settings,
+) *admission.Pacer {
+	var pacer *admission.Pacer
+	if fileSSTSinkElasticCPUControlEnabled.Get(&settings.SV) {
+		tenantID, ok := roachpb.ClientTenantFromContext(ctx)
+		if !ok {
+			tenantID = roachpb.SystemTenantID
+		}
+		pacer = factory.NewPacer(
+			100*time.Millisecond,
+			admission.WorkInfo{
+				TenantID:        tenantID,
+				Priority:        admissionpb.BulkNormalPri,
+				CreateTime:      timeutil.Now().UnixNano(),
+				BypassAdmission: false,
+			},
+		)
+	}
+	return pacer
+}
+
 func init() {
 	rowexec.NewBackupDataProcessor = newBackupDataProcessor
 }

pkg/backup/backupsink/sst_sink_key_writer.go

@@ -15,7 +15,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/storage"
-	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
 )
@@ -32,14 +31,12 @@ type SSTSinkKeyWriter struct {
 	targetFileSize int64
 }
 
-func MakeSSTSinkKeyWriter(
-	conf SSTSinkConf, dest cloud.ExternalStorage, pacer *admission.Pacer,
-) (*SSTSinkKeyWriter, error) {
+func MakeSSTSinkKeyWriter(conf SSTSinkConf, dest cloud.ExternalStorage) (*SSTSinkKeyWriter, error) {
 	if conf.ElideMode == execinfrapb.ElidePrefix_None {
 		return nil, errors.New("KeyWriter does not support ElidePrefix_None")
 	}
 	return &SSTSinkKeyWriter{
-		FileSSTSink:    *MakeFileSSTSink(conf, dest, pacer),
+		FileSSTSink:    *MakeFileSSTSink(conf, dest, nil),
 		targetFileSize: targetFileSize.Get(conf.Settings),
 	}, nil
 }

pkg/backup/backupsink/sst_sink_key_writer_test.go

@@ -485,7 +485,7 @@ func sstSinkKeyWriterTestSetup(
 	t *testing.T, settings *cluster.Settings, elideMode execinfrapb.ElidePrefix,
 ) (*SSTSinkKeyWriter, cloud.ExternalStorage) {
 	conf, store := sinkTestSetup(t, settings, elideMode)
-	sink, err := MakeSSTSinkKeyWriter(conf, store, nil /* pacer */)
+	sink, err := MakeSSTSinkKeyWriter(conf, store)
 	require.NoError(t, err)
 	return sink, store
 }

pkg/backup/compaction_processor.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/storage"
+	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -251,7 +252,7 @@ func (p *compactBackupsProcessor) processSpanEntries(
 		Settings:  &execCfg.Settings.SV,
 		ElideMode: p.spec.ElideMode,
 	}
-	sink, err := backupsink.MakeSSTSinkKeyWriter(sinkConf, store, nil)
+	sink, err := backupsink.MakeSSTSinkKeyWriter(sinkConf, store)
 	if err != nil {
 		return err
 	}
@@ -261,6 +262,11 @@ func (p *compactBackupsProcessor) processSpanEntries(
 			logClose(ctx, sink, "SST sink")
 		}
 	}()
+	pacer := newBackupPacer(
+		ctx, p.FlowCtx.Cfg.AdmissionPacerFactory, p.FlowCtx.Cfg.Settings,
+	)
+	// It is safe to close a nil pacer.
+	defer pacer.Close()
 
 	for {
 		select {
@@ -280,7 +286,7 @@ func (p *compactBackupsProcessor) processSpanEntries(
 				return errors.Wrap(err, "opening SSTs")
 			}
 
-			if err := compactSpanEntry(ctx, sstIter, sink); err != nil {
+			if err := compactSpanEntry(ctx, sstIter, sink, pacer); err != nil {
 				return errors.Wrap(err, "compacting span entry")
 			}
 		}
@@ -385,7 +391,7 @@ func openSSTs(
 }
 
 func compactSpanEntry(
-	ctx context.Context, sstIter mergedSST, sink *backupsink.SSTSinkKeyWriter,
+	ctx context.Context, sstIter mergedSST, sink *backupsink.SSTSinkKeyWriter, pacer *admission.Pacer,
 ) error {
 	defer sstIter.cleanup()
 	entry := sstIter.entry
@@ -404,6 +410,9 @@ func compactSpanEntry(
 	scratch = append(scratch, prefix...)
 	iter := sstIter.iter
 	for iter.SeekGE(trimmedStart); ; iter.NextKey() {
+		if err := pacer.Pace(ctx); err != nil {
+			return err
+		}
 		var key storage.MVCCKey
 		if ok, err := iter.Valid(); err != nil {
 			return err