Merge #151030

151030: *: treat ac pacer errs as errors r=dt a=dt

The AC CPU pacer.Pace() only returns a non-nil error in situations where the caller should stop processing paced work, such as if the context is cancelled, meaning the caller of the pacer should generally return its non-nil err.

Previously many callers were treating errors from Pace() as failures to pace, but a failure to pace was simply something to note to a log before continuing to run the (now unpaced) operation.

This changes that log-and-continue handling of Pace errors across users of the pacer to the more standard 'return err' handling.

Release note: none.
Epic: none.

Co-authored-by: David Taylor <[email protected]>

Author: craig[bot]

pkg/kv/bulk/cpu_pacer.go

@@ -18,27 +18,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 )
 
-// CPUPacer wraps an admission.Pacer for use in bulk operations where any errors
-// pacing can just be logged (at most once a minute).
-type CPUPacer struct {
-	pacer    *admission.Pacer
-	logEvery *log.EveryN
-}
-
-func (p *CPUPacer) Pace(ctx context.Context) {
-	if err := p.pacer.Pace(ctx); err != nil {
-		if p.logEvery.ShouldLog() {
-			// TODO(dt): consider just returning this error/eliminating this wrapper,
-			// and making callers bubble up this error if it is only ever a ctx cancel
-			// error that the caller should be bubbling up anyway.
-			log.Warningf(ctx, "failed to pace SST batcher: %v", err)
-		}
-	}
-}
-func (p *CPUPacer) Close() {
-	p.pacer.Close()
-}
-
 var cpuPacerRequestDuration = settings.RegisterDurationSetting(
 	settings.ApplicationLevel,
 	"bulkio.elastic_cpu_control.request_duration",
@@ -48,24 +27,21 @@ var cpuPacerRequestDuration = settings.RegisterDurationSetting(
 
 // NewCPUPacer creates a new AC pacer for SST batcher. It may return an empty
 // Pacer which noops if pacing is disabled or its arguments are nil.
-func NewCPUPacer(ctx context.Context, db *kv.DB, setting *settings.BoolSetting) CPUPacer {
+func NewCPUPacer(ctx context.Context, db *kv.DB, setting *settings.BoolSetting) *admission.Pacer {
 	if db == nil || db.AdmissionPacerFactory == nil || !setting.Get(db.SettingsValues()) {
 		log.Infof(ctx, "admission control is not configured to pace bulk ingestion")
-		return CPUPacer{}
+		return nil
 	}
 	tenantID, ok := roachpb.ClientTenantFromContext(ctx)
 	if !ok {
 		tenantID = roachpb.SystemTenantID
 	}
-	every := log.Every(time.Minute)
-	return CPUPacer{pacer: db.AdmissionPacerFactory.NewPacer(
+	return db.AdmissionPacerFactory.NewPacer(
 		cpuPacerRequestDuration.Get(db.SettingsValues()),
 		admission.WorkInfo{
 			TenantID:        tenantID,
 			Priority:        admissionpb.BulkNormalPri,
 			CreateTime:      timeutil.Now().UnixNano(),
 			BypassAdmission: false,
-		}),
-		logEvery: &every,
-	}
+		})
 }

pkg/kv/bulk/sst_batcher.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
+	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb"
 	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
@@ -220,7 +221,7 @@ type SSTBatcher struct {
 	disallowShadowingBelow hlc.Timestamp
 
 	// pacer for admission control during SST ingestion
-	pacer CPUPacer
+	pacer *admission.Pacer
 
 	// skips duplicate keys (iff they are buffered together). This is true when
 	// used to backfill an inverted index. An array in JSONB with multiple values
@@ -444,7 +445,9 @@ func (b *SSTBatcher) AddMVCCKeyLDR(ctx context.Context, key storage.MVCCKey, val
 // Keys must be added in order.
 func (b *SSTBatcher) AddMVCCKey(ctx context.Context, key storage.MVCCKey, value []byte) error {
 	// Pace based on admission control before adding the key.
-	b.pacer.Pace(ctx)
+	if err := b.pacer.Pace(ctx); err != nil {
+		return err
+	}
 
 	if len(b.batch.endKey) > 0 && bytes.Equal(b.batch.endKey, key.Key) {
 		if b.ingestAll && key.Timestamp.Equal(b.batch.endTimestamp) {

pkg/kv/kvserver/rangefeed/catchup_scan.go

@@ -8,7 +8,6 @@ package rangefeed
 import (
 	"bytes"
 	"context"
-	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -18,7 +17,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/bufalloc"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
-	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
 )
@@ -191,7 +189,6 @@ func (i *CatchUpIterator) CatchUpScan(
 	var meta enginepb.MVCCMetadata
 	i.SeekGE(storage.MVCCKey{Key: i.span.Key})
 
-	every := log.Every(100 * time.Millisecond)
 	for {
 		if ok, err := i.Valid(); err != nil {
 			return err
@@ -200,11 +197,7 @@ func (i *CatchUpIterator) CatchUpScan(
 		}
 
 		if err := i.pacer.Pace(ctx); err != nil {
-			// We're unable to pace things automatically -- shout loudly
-			// semi-infrequently but don't fail the rangefeed itself.
-			if every.ShouldLog() {
-				log.Errorf(ctx, "automatic pacing: %v", err)
-			}
+			return err
 		}
 
 		// Emit any new MVCC range tombstones when their start key is encountered.

pkg/sql/importer/read_import_base.go

@@ -584,7 +584,9 @@ func runParallelImport(
 		var numSkipped int64
 		var count int64
 		for producer.Scan() {
-			pacer.Pace(ctx)
+			if err := pacer.Pace(ctx); err != nil {
+				return err
+			}
 			// Skip rows if needed.
 			count++
 			if count <= fileCtx.skip {
@@ -700,7 +702,10 @@ func (p *parallelImporter) importWorker(
 		for batchIdx, record := range batch.data {
 			rowNum = batch.startPos + int64(batchIdx)
 			// Pace the admission control before processing each row.
-			pacer.Pace(ctx)
+			if err := pacer.Pace(ctx); err != nil {
+				return err
+			}
+
 			if err := consumer.FillDatums(ctx, record, rowNum, conv); err != nil {
 				if err = handleCorruptRow(ctx, fileCtx, err); err != nil {
 					return err

pkg/sql/importer/read_import_workload.go

@@ -275,7 +275,9 @@ func (w *WorkloadKVConverter) Worker(
 		a = a.Truncate()
 		w.rows.FillBatch(batchIdx, cb, &a)
 		for rowIdx, numRows := 0, cb.Length(); rowIdx < numRows; rowIdx++ {
-			pacer.Pace(ctx)
+			if err := pacer.Pace(ctx); err != nil {
+				return err
+			}
 			for colIdx, col := range cb.ColVecs() {
 				// TODO(dan): This does a type switch once per-datum. Reduce this to
 				// a one-time switch per column.

pkg/sql/row/kv_batch_fetcher.go

@@ -55,8 +55,6 @@ var defaultKVBatchSize = rowinfra.KeyLimit(metamorphic.ConstantWithTestValue(
 	1,                                   /* metamorphicValue */
 ))
 
-var logAdmissionPacerErr = log.Every(100 * time.Millisecond)
-
 // elasticCPUDurationPerLowPriReadResponse controls how many CPU tokens are allotted
 // each time we seek admission for response handling during internally submitted
 // low priority reads (like row-level TTL selects).
@@ -753,12 +751,7 @@ func (f *txnKVFetcher) maybeAdmitBatchResponse(ctx context.Context, br *kvpb.Bat
 		// to ensure that they have local plans with a single TableReader
 		// processor in multi-node clusters.
 		if err := f.admissionPacer.Pace(ctx); err != nil {
-			// We're unable to pace things automatically -- shout loudly
-			// semi-infrequently but don't fail the kv fetcher itself. At
-			// worst we'd be over-admitting.
-			if logAdmissionPacerErr.ShouldLog() {
-				log.Errorf(ctx, "automatic pacing: %v", err)
-			}
+			return err
 		}
 	} else if f.responseAdmissionQ != nil {
 		responseAdmission := admission.WorkInfo{

pkg/sql/rowexec/indexbackfiller.go

@@ -345,7 +345,9 @@ func (ib *indexBackfiller) ingestIndexEntries(
 		for indexBatch := range indexEntryCh {
 			for _, indexEntry := range indexBatch.indexEntries {
 				// Pace the admission control before processing each index entry.
-				pacer.Pace(ctx)
+				if err := pacer.Pace(ctx); err != nil {
+					return err
+				}
 
 				// If there is at least one vector index being written, we need to check to see
 				// if this IndexEntry is going to a vector index and then re-encode it for that

pkg/util/admission/elastic_cpu_work_queue.go

@@ -66,6 +66,7 @@ func makeElasticCPUWorkQueue(
 }
 
 // Admit is called when requesting admission for elastic CPU work.
+// Non-nil errors are returned only if the context is canceled.
 func (e *ElasticCPUWorkQueue) Admit(
 	ctx context.Context, duration time.Duration, info WorkInfo,
 ) (*ElasticCPUWorkHandle, error) {

pkg/util/admission/pacer.go

@@ -23,7 +23,9 @@ type Pacer struct {
 	cur *ElasticCPUWorkHandle
 }
 
-// Pace is part of the Pacer interface.
+// Pace will block as needed to pace work that calls it as configured. It is
+// intended to be called in a tight loop, and will attempt to minimize per-call
+// overhead. Non-nil errors are returned only if the context is canceled.
 func (p *Pacer) Pace(ctx context.Context) error {
 	if p == nil {
 		return nil