Merge cockroachdb#157210

craig[bot] · stevendanna · craig[bot] · commit a59328e4e34c · 2025-11-14T09:53:32.000Z
157210: util,log: make util.EveryN generic r=yuzefovich a=stevendanna

This changes util.EveryN to a generic function and then

- modifies log.EveryN to allow use a crtime.Mono instead of a time.Time.
- updates most callers that were easy to update to use util.EveryMono

We might consider removing the time.Time version of EveryN, but it is still useful in a couple of places.

Epic: none
Release note: None

Co-authored-by: Steven Danna &lt;danna@cockroachlabs.com&gt;
diff --git a/pkg/backup/BUILD.bazel b/pkg/backup/BUILD.bazel
@@ -157,6 +157,7 @@ go_library(
         "//pkg/util/tracing",
         "//pkg/util/tracing/tracingpb",
         "//pkg/util/uuid",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_pebble//:pebble",
diff --git a/pkg/backup/restore_job.go b/pkg/backup/restore_job.go
@@ -79,6 +79,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
 )
@@ -215,7 +216,7 @@ func restoreWithRetry(
 	// dying), so if we receive a retryable error, re-plan and retry the restore.
 	retryOpts, progThreshold := getRetryOptionsAndProgressThreshold(execCtx)
 	logRate := restoreRetryLogRate.Get(&execCtx.ExecCfg().Settings.SV)
-	logThrottler := util.Every(logRate)
+	logThrottler := util.EveryMono(logRate)
 	var (
 		res                roachpb.RowCount
 		err                error
@@ -255,7 +256,7 @@ func restoreWithRetry(
 
 		log.Dev.Warningf(ctx, "encountered retryable error: %+v", err)
 
-		if logThrottler.ShouldProcess(timeutil.Now()) {
+		if logThrottler.ShouldProcess(crtime.NowMono()) {
 			// We throttle the logging of errors to the jobs messages table to avoid
 			// flooding the table during the hot loop of a retry.
 			if err := execCtx.ExecCfg().InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
diff --git a/pkg/ccl/changefeedccl/cdcevent/BUILD.bazel b/pkg/ccl/changefeedccl/cdcevent/BUILD.bazel
@@ -43,6 +43,7 @@ go_library(
         "//pkg/util/json",
         "//pkg/util/log",
         "//pkg/util/protoutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
     ],
diff --git a/pkg/ccl/changefeedccl/cdcevent/rowfetcher_cache.go b/pkg/ccl/changefeedccl/cdcevent/rowfetcher_cache.go
@@ -28,6 +28,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
 
@@ -318,7 +319,7 @@ func (c *rowFetcherCache) RowFetcherForColumnFamily(
 			Alloc:             &c.a,
 			Spec:              &spec,
 			TraceKV:           c.rfArgs.traceKV,
-			TraceKVEvery:      &util.EveryN{N: c.rfArgs.traceKVLogFrequency},
+			TraceKVEvery:      &util.EveryN[crtime.Mono]{N: c.rfArgs.traceKVLogFrequency},
 		},
 	); err != nil {
 		return nil, nil, err
diff --git a/pkg/ccl/changefeedccl/changefeed_processors.go b/pkg/ccl/changefeedccl/changefeed_processors.go
@@ -1116,7 +1116,7 @@ const (
 )
 
 // slowLogEveryN rate-limits the logging of slow spans
-var slowLogEveryN = log.Every(slowSpanMaxFrequency)
+var slowLogEveryN = util.Every(slowSpanMaxFrequency)
 
 // jobState encapsulates changefeed job state.
 type jobState struct {
@@ -2434,7 +2434,7 @@ type saveRateConfig struct {
 // duration it takes to save progress.
 type saveRateLimiter struct {
 	config     saveRateConfig
-	warnEveryN util.EveryN
+	warnEveryN util.EveryN[time.Time]
 
 	clock timeutil.TimeSource
 
diff --git a/pkg/cmd/roachtest/roachtestutil/utils.go b/pkg/cmd/roachtest/roachtestutil/utils.go
@@ -59,7 +59,7 @@ func SetDefaultAdminUIPort(c cluster.Cluster, opts *install.StartOpts) {
 // recently a given log message has been emitted so that it can determine
 // whether it's worth logging again.
 type EveryN struct {
-	util.EveryN
+	util.EveryN[time.Time]
 }
 
 // Every is a convenience constructor for an EveryN object that allows a log
diff --git a/pkg/kv/kvserver/raft_log_queue.go b/pkg/kv/kvserver/raft_log_queue.go
@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
@@ -147,7 +148,7 @@ type raftLogQueue struct {
 	*baseQueue
 	db *kv.DB
 
-	logSnapshots util.EveryN
+	logSnapshots util.EveryN[crtime.Mono]
 }
 
 var _ queueImpl = &raftLogQueue{}
@@ -162,7 +163,7 @@ var _ queueImpl = &raftLogQueue{}
 func newRaftLogQueue(store *Store, db *kv.DB) *raftLogQueue {
 	rlq := &raftLogQueue{
 		db:           db,
-		logSnapshots: util.Every(10 * time.Second),
+		logSnapshots: util.EveryMono(10 * time.Second),
 	}
 	rlq.baseQueue = newBaseQueue(
 		"raftlog", rlq, store,
@@ -689,7 +690,7 @@ func (rlq *raftLogQueue) process(
 		return false, nil
 	}
 
-	if n := decision.NumNewRaftSnapshots(); log.V(1) || n > 0 && rlq.logSnapshots.ShouldProcess(timeutil.Now()) {
+	if n := decision.NumNewRaftSnapshots(); log.V(1) || n > 0 && rlq.logSnapshots.ShouldProcess(crtime.NowMono()) {
 		log.KvExec.Infof(ctx, "%v", redact.Safe(decision.String()))
 		log.KvDistribution.Infof(ctx, "%v", redact.Safe(decision.String()))
 	} else {
diff --git a/pkg/kv/kvserver/replica.go b/pkg/kv/kvserver/replica.go
@@ -59,6 +59,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 	"github.com/kr/pretty"
@@ -1056,7 +1057,7 @@ type Replica struct {
 	// information and without explicit throttling some replicas will offer once
 	// per applied Raft command, which is silly and also clogs up the queues'
 	// semaphores.
-	splitQueueThrottle, mergeQueueThrottle util.EveryN
+	splitQueueThrottle, mergeQueueThrottle util.EveryN[crtime.Mono]
 
 	// loadBasedSplitter keeps information about load-based splitting.
 	loadBasedSplitter split.Decider
diff --git a/pkg/kv/kvserver/replica_app_batch.go b/pkg/kv/kvserver/replica_app_batch.go
@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logcrash"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
@@ -688,7 +689,7 @@ func (b *replicaAppBatch) ApplyToStateMachine(ctx context.Context) error {
 	// Record the number of keys written to the replica.
 	b.r.loadStats.RecordWriteKeys(float64(b.ab.numMutations))
 
-	now := timeutil.Now()
+	now := crtime.NowMono()
 	if needsSplitBySize && r.splitQueueThrottle.ShouldProcess(now) {
 		r.store.splitQueue.MaybeAddAsync(ctx, r, r.store.Clock().NowAsClockTimestamp())
 	}
diff --git a/pkg/kv/kvserver/replica_init.go b/pkg/kv/kvserver/replica_init.go
@@ -242,8 +242,8 @@ func newUninitializedReplicaWithoutRaftGroup(store *Store, id roachpb.FullReplic
 			store.TestingKnobs().DisableSyncLogWriteToss,
 	}
 
-	r.splitQueueThrottle = util.Every(splitQueueThrottleDuration)
-	r.mergeQueueThrottle = util.Every(mergeQueueThrottleDuration)
+	r.splitQueueThrottle = util.EveryMono(splitQueueThrottleDuration)
+	r.mergeQueueThrottle = util.EveryMono(mergeQueueThrottleDuration)
 
 	onTrip := func() {
 		telemetry.Inc(telemetryTripAsync)
diff --git a/pkg/spanconfig/spanconfigjob/BUILD.bazel b/pkg/spanconfig/spanconfigjob/BUILD.bazel
@@ -18,6 +18,7 @@ go_library(
         "//pkg/util/retry",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
     ],
 )
diff --git a/pkg/spanconfig/spanconfigjob/job.go b/pkg/spanconfig/spanconfigjob/job.go
@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
 
@@ -101,14 +102,14 @@ func (r *resumer) Resume(ctx context.Context, execCtxI interface{}) (jobErr erro
 	settingValues := &execCtx.ExecCfg().Settings.SV
 	persistCheckpointsMu := struct {
 		syncutil.Mutex
-		util.EveryN
+		util.EveryN[crtime.Mono]
 	}{}
-	persistCheckpointsMu.EveryN = util.Every(ReconciliationJobCheckpointInterval.Get(settingValues))
+	persistCheckpointsMu.EveryN = util.EveryMono(ReconciliationJobCheckpointInterval.Get(settingValues))
 
 	ReconciliationJobCheckpointInterval.SetOnChange(settingValues, func(ctx context.Context) {
 		persistCheckpointsMu.Lock()
 		defer persistCheckpointsMu.Unlock()
-		persistCheckpointsMu.EveryN = util.Every(ReconciliationJobCheckpointInterval.Get(settingValues))
+		persistCheckpointsMu.EveryN = util.EveryMono(ReconciliationJobCheckpointInterval.Get(settingValues))
 	})
 
 	checkpointingDisabled := false
@@ -152,7 +153,7 @@ func (r *resumer) Resume(ctx context.Context, execCtxI interface{}) (jobErr erro
 			shouldPersistCheckpoint := func() bool {
 				persistCheckpointsMu.Lock()
 				defer persistCheckpointsMu.Unlock()
-				return persistCheckpointsMu.ShouldProcess(timeutil.Now())
+				return persistCheckpointsMu.ShouldProcess(crtime.NowMono())
 			}()
 
 			if !shouldPersistCheckpoint {
diff --git a/pkg/spanconfig/spanconfigsqlwatcher/BUILD.bazel b/pkg/spanconfig/spanconfigsqlwatcher/BUILD.bazel
@@ -35,7 +35,7 @@ go_library(
         "//pkg/util/protoutil",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
-        "//pkg/util/timeutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
     ],
 )
diff --git a/pkg/spanconfig/spanconfigsqlwatcher/sqlwatcher.go b/pkg/spanconfig/spanconfigsqlwatcher/sqlwatcher.go
@@ -26,7 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logcrash"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
-	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
 
@@ -151,7 +151,7 @@ func (s *SQLWatcher) watch(
 	}
 	defer ptsRF.Close()
 
-	checkpointNoops := util.Every(s.checkpointNoopsEvery)
+	checkpointNoops := util.EveryMono(s.checkpointNoopsEvery)
 	for {
 		select {
 		case <-ctx.Done():
@@ -166,7 +166,7 @@ func (s *SQLWatcher) watch(
 				return err
 			}
 			if len(sqlUpdates) == 0 &&
-				(!checkpointNoops.ShouldProcess(timeutil.Now()) || s.knobs.SQLWatcherSkipNoopCheckpoints) {
+				(!checkpointNoops.ShouldProcess(crtime.NowMono()) || s.knobs.SQLWatcherSkipNoopCheckpoints) {
 				continue
 			}
 			if err := handler(ctx, sqlUpdates, combinedFrontierTS); err != nil {
diff --git a/pkg/sql/row/BUILD.bazel b/pkg/sql/row/BUILD.bazel
@@ -94,6 +94,7 @@ go_library(
         "//pkg/util/timeutil",
         "//pkg/util/unique",
         "//pkg/util/uuid",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_lib_pq//oid",
     ],
diff --git a/pkg/sql/row/fetcher.go b/pkg/sql/row/fetcher.go
@@ -39,6 +39,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/lib/pq/oid"
 )
@@ -339,7 +340,7 @@ type FetcherInitArgs struct {
 	TraceKV bool
 	// TraceKVEvery controls how often KVs are sampled for logging with traceKV
 	// enabled.
-	TraceKVEvery               *util.EveryN
+	TraceKVEvery               *util.EveryN[crtime.Mono]
 	ForceProductionKVBatchSize bool
 	// SpansCanOverlap indicates whether the spans in a given batch can overlap
 	// with one another. If it is true, spans that correspond to the same row must
@@ -1196,7 +1197,7 @@ func (rf *Fetcher) NextRow(ctx context.Context) (row rowenc.EncDatumRow, spanID
 		// log.EveryN will always print under verbosity level 2.
 		// The caller may choose to set it to avoid logging
 		// too many rows. If unset, we log every KV.
-		if rf.args.TraceKV && (rf.args.TraceKVEvery == nil || rf.args.TraceKVEvery.ShouldProcess(timeutil.Now())) {
+		if rf.args.TraceKV && (rf.args.TraceKVEvery == nil || rf.args.TraceKVEvery.ShouldProcess(crtime.NowMono())) {
 			log.VEventf(ctx, TraceKVVerbosity, "fetched: %s -> %s", prettyKey, prettyVal)
 		}
 
diff --git a/pkg/sql/rowexec/BUILD.bazel b/pkg/sql/rowexec/BUILD.bazel
@@ -118,6 +118,7 @@ go_library(
         "//pkg/util/tracing/tracingpb",
         "//pkg/util/vector",
         "@com_github_axiomhq_hyperloglog//:hyperloglog",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_redact//:redact",
diff --git a/pkg/sql/rowexec/sample_aggregator.go b/pkg/sql/rowexec/sample_aggregator.go
@@ -30,9 +30,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/intsets"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
-	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
 
@@ -245,14 +245,14 @@ func (s *sampleAggregator) mainLoop(
 	}
 
 	var rowsProcessed uint64
-	progressUpdates := util.Every(SampleAggregatorProgressInterval)
+	progressUpdates := util.EveryMono(SampleAggregatorProgressInterval)
 	var da tree.DatumAlloc
 	for {
 		row, meta := s.input.Next()
 		if meta != nil {
 			if meta.SamplerProgress != nil {
 				rowsProcessed += meta.SamplerProgress.RowsProcessed
-				if progressUpdates.ShouldProcess(timeutil.Now()) {
+				if progressUpdates.ShouldProcess(crtime.NowMono()) {
 					// Periodically report fraction progressed and check that the job has
 					// not been paused or canceled.
 					var fractionCompleted float32
diff --git a/pkg/sql/unsafesql/BUILD.bazel b/pkg/sql/unsafesql/BUILD.bazel
@@ -13,7 +13,7 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/log/eventpb",
         "//pkg/util/log/severity",
-        "//pkg/util/timeutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_redact//:redact",
     ],
 )
diff --git a/pkg/sql/unsafesql/unsafesql.go b/pkg/sql/unsafesql/unsafesql.go
@@ -16,7 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
 	"github.com/cockroachdb/cockroach/pkg/util/log/severity"
-	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/redact"
 )
 
@@ -64,14 +64,14 @@ func CheckInternalsAccess(
 	// If an override is set, allow access to this virtual table.
 	if allowUnsafe {
 		// Log this access to the SENSITIVE_ACCESS channel since the override condition bypassed normal access controls.
-		if accessedLogLimiter.ShouldProcess(timeutil.Now()) {
+		if accessedLogLimiter.ShouldProcess(crtime.NowMono()) {
 			log.StructuredEvent(ctx, severity.WARNING, &eventpb.UnsafeInternalsAccessed{Query: q})
 		}
 		return nil
 	}
 
 	// Log this access to the SENSITIVE_ACCESS channel to show where failing internals accesses are happening.
-	if deniedLogLimiter.ShouldProcess(timeutil.Now()) {
+	if deniedLogLimiter.ShouldProcess(crtime.NowMono()) {
 		log.StructuredEvent(ctx, severity.WARNING, &eventpb.UnsafeInternalsDenied{Query: q})
 	}
 
diff --git a/pkg/util/BUILD.bazel b/pkg/util/BUILD.bazel
@@ -27,6 +27,7 @@ go_library(
     deps = [
         "//pkg/util/netutil/addr",
         "//pkg/util/syncutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
     ] + select({
@@ -56,6 +57,7 @@ go_test(
     deps = [
         "//pkg/util/randutil",
         "//pkg/util/timeutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
     ],
diff --git a/pkg/util/admission/BUILD.bazel b/pkg/util/admission/BUILD.bazel
@@ -47,6 +47,7 @@ go_library(
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_pebble//:pebble",
diff --git a/pkg/util/admission/elastic_cpu_granter.go b/pkg/util/admission/elastic_cpu_granter.go
diff --git a/pkg/util/every_n.go b/pkg/util/every_n.go
diff --git a/pkg/util/every_n_test.go b/pkg/util/every_n_test.go
diff --git a/pkg/util/log/BUILD.bazel b/pkg/util/log/BUILD.bazel
diff --git a/pkg/util/log/every_n.go b/pkg/util/log/every_n.go

Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,7 @@ func SetDefaultAdminUIPort(c cluster.Cluster, opts *install.StartOpts) {`
`59`	`59`	`// recently a given log message has been emitted so that it can determine`
`60`	`60`	`// whether it's worth logging again.`
`61`	`61`	`type EveryN struct {`
`62`		`- util.EveryN`
	`62`	`+ util.EveryN[time.Time]`
`63`	`63`	`}`
`64`	`64`
`65`	`65`	`// Every is a convenience constructor for an EveryN object that allows a log`
Original file line number	Diff line number	Diff line change
`@@ -242,8 +242,8 @@ func newUninitializedReplicaWithoutRaftGroup(store *Store, id roachpb.FullReplic`
`242`	`242`	`store.TestingKnobs().DisableSyncLogWriteToss,`
`243`	`243`	`}`
`244`	`244`
`245`		`- r.splitQueueThrottle = util.Every(splitQueueThrottleDuration)`
`246`		`- r.mergeQueueThrottle = util.Every(mergeQueueThrottleDuration)`
	`245`	`+ r.splitQueueThrottle = util.EveryMono(splitQueueThrottleDuration)`
	`246`	`+ r.mergeQueueThrottle = util.EveryMono(mergeQueueThrottleDuration)`
`247`	`247`
`248`	`248`	`onTrip := func() {`
`249`	`249`	`telemetry.Inc(telemetryTripAsync)`